refactor(dapi): rewrite dapi in Rust as rs-dapi #2716
Conversation
WalkthroughAdds a new Rust rs-dapi service and rs-dash-event-bus crate; integrates rs-dapi into workspace, Docker, CI, and dashmate; implements gRPC + JSON‑RPC servers, clients, streaming (ZMQ), caching, logging, metrics, config migrations, many manifest/version bumps, docs, examples, and tests. Changes
Sequence Diagram(s)%%{init: {"themeVariables": {"primaryColor":"#8FB9A8","secondaryColor":"#DCEFE0"}}}%%
sequenceDiagram
participant Client
participant JSONRPC as JSON‑RPC HTTP
participant Translator as JsonRpcTranslator
participant Platform as PlatformServiceImpl
participant Drive as DriveClient
participant Core as CoreClient
participant Tender as TenderdashClient
Client->>JSONRPC: POST JSON‑RPC request
JSONRPC->>Translator: translate_request()
Translator->>Platform: JsonRpcCall
alt needs Drive
Platform->>Drive: get_*/call (cached)
Drive-->>Platform: response
end
alt needs Core
Platform->>Core: get_*/call (cached)
Core-->>Platform: response
end
alt needs Tenderdash
Platform->>Tender: RPC / WS
Tender-->>Platform: response/event
end
Platform-->>Translator: result
Translator->>JSONRPC: JSON‑RPC response
JSONRPC-->>Client: HTTP response
%%{init: {"themeVariables": {"primaryColor":"#F2C1C1","noteBackground":"#FFFBE6"}}}%%
sequenceDiagram
participant CoreNode as Dash Core (ZMQ)
participant ZMQ as ZmqListener
participant Bus as SubscriberManager/EventBus
participant Stream as StreamingService
participant Client
CoreNode->>ZMQ: publish raw events
ZMQ->>Bus: broadcast(StreamingEvent)
Client->>Stream: subscribe(filter)
Stream->>Bus: add_subscription(filter)
Bus-->>Client: deliver matching events
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120+ minutes Possibly related PRs
Suggested reviewers
Poem
Pre-merge checks and finishing touches✅ Passed checks (3 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
✅ gRPC Query Coverage Report |
lklimek
force-pushed
the
refactor/dapi-rewrite-to-rust
branch
from
d7a4c79 to
84dc6d6
Compare
|
✅ DashSDKFFI.xcframework built for this PR.
SwiftPM (host the zip at a stable URL, then use): .binaryTarget(
name: "DashSDKFFI",
url: "https://your.cdn.example/DashSDKFFI.xcframework.zip",
checksum: "1f048919e0c4b263e59f94e5f43e480476f36b2bdf342f71e1f9833f6315c824"
)Xcode manual integration:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (2)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (2)
331-334: Ack: renamed to is_running() as suggested earlier.Name matches semantics (cancellation token). Good.
529-554: Replace ping-based health check with last-message lag detection.Subscribing/unsubscribing "ping" doesn’t validate socket liveness and can be misleading; track time since last received message and cancel when stale (e.g., >60s). Based on learnings.
Apply within this method:
- /// Event that happens every ten seconds to check connection status - async fn tick_event_10s(&mut self) { - // Health check of zmq connection - // This is a hack to ensure the connection is alive, as the monitor fails to notify us about disconnects - let current_status = self.socket.subscribe("ping").await.is_ok(); - // Unsubscribe immediately to avoid resource waste - self.socket - .unsubscribe("ping") - .await - .inspect_err(|e| { - debug!(error = %e, "Error unsubscribing from ping topic during health check"); - }) - .ok(); - - // If the status changed, log it - let previous_status = self.connected.swap(current_status, Ordering::SeqCst); - if current_status != previous_status { - if current_status { - debug!("ZMQ connection recovered"); - } else { - debug!("ZMQ connection is lost, connection will be restarted"); - // disconnect the socket - self.cancel.cancel(); - } - } - } + /// Event that happens every ten seconds to check connection status + async fn tick_event_10s(&mut self) { + let silent_for = self.last_msg.elapsed(); + if silent_for > Duration::from_secs(60) { + debug!(?silent_for, "No ZMQ messages for too long; restarting connection"); + self.connected.store(false, Ordering::SeqCst); + self.cancel.cancel(); + } + }And add supporting changes elsewhere:
Add to imports:
use std::time::Instant;Extend struct:
struct ZmqDispatcher { socket: SubSocket, zmq_tx: mpsc::Sender<ZmqMessage>, cancel: CancellationToken, connected: Arc<AtomicBool>, last_msg: Instant, }Initialize in start_dispatcher:
last_msg: Instant::now(),Update on receive success inside dispatcher_worker:
Ok(msg) => { self.last_msg = Instant::now(); if let Err(e) = self.zmq_tx.send(msg).await { /* unchanged */ } }
🧹 Nitpick comments (3)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (3)
309-321: Simplify spawn; remove unreachable branch after awaiting an infinite task.zmq_listener_task loops forever; the if-let after await never runs. Pass the task directly to with_cancel.
- tokio::task::spawn(with_cancel(cancel.clone(), async move { - // we use child token so that cancelling threads started inside zmq_listener_task - // does not cancel the zmq_listener_task itself, as it needs to restart the - // connection if it fails - if let Err(e) = - Self::zmq_listener_task(zmq_uri, topics, sender, cancel.child_token()).await - { - debug!(error = %e, "ZMQ listener task error"); - // we cancel parent task to stop all spawned threads - cancel.cancel(); - } - Err::<(), _>(DapiError::ConnectionClosed) - })); + tokio::task::spawn(with_cancel( + cancel.clone(), + Self::zmq_listener_task(zmq_uri, topics, sender, cancel.child_token()), + ));
326-330: Make subscribe() synchronous.No awaits; remove async for simpler API.
- pub async fn subscribe(&self) -> DAPIResult<broadcast::Receiver<ZmqEvent>> { + pub fn subscribe(&self) -> DAPIResult<broadcast::Receiver<ZmqEvent>> { Ok(self.event_sender.subscribe()) }
137-141: Use a connection-oriented error for timeouts.Configuration isn’t the root cause on timeout; prefer a connection/timeout error variant.
- .map_err(|_| DapiError::Configuration("Connection timeout".to_string()))? + .map_err(|_| DapiError::ConnectionClosed)?If you have a dedicated timeout variant, use it instead.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs(1 hunks)
🧰 Additional context used
📓 Path-based instructions (2)
**/*.rs
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.rs: Format Rust code with cargo fmt
Run Clippy linter for Rust code
Files:
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs
packages/rs-*/**/*.rs
📄 CodeRabbit inference engine (AGENTS.md)
packages/rs-*/**/*.rs: Use rustfmt defaults for Rust code (format via cargo fmt --all)
Keep Rust code clippy-clean (cargo clippy)
Name Rust modules in snake_case
Use PascalCase for Rust types
Use SCREAMING_SNAKE_CASE for Rust constants
Files:
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs
🧠 Learnings (1)
📓 Common learnings
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/dashmate/src/test/constants/services.js:4-4
Timestamp: 2025-10-15T14:45:30.856Z
Learning: In the dashmate codebase (packages/dashmate), during the DAPI Rust migration (rs-dapi), the old service keys `dapi_api` and `dapi_core_streams` are intentionally kept in `generateEnvsFactory.js` for backward compatibility even though the test constants in `src/test/constants/services.js` have been updated to use `rs_dapi`. These deprecated keys will be removed in a future PR after the transition is complete.
Learnt from: shumkov
PR: dashpay/platform#2270
File: packages/dapi/lib/grpcServer/handlers/platform/broadcastStateTransitionHandlerFactory.js:69-125
Timestamp: 2024-10-24T05:07:35.892Z
Learning: Future development plans involve rewriting DAPI into Rust, as the rest of the project has already migrated.
🧬 Code graph analysis (1)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (1)
packages/rs-dapi/src/services/streaming_service/mod.rs (2)
new(48-67)summarize_zmq_event(445-486)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (166)
- GitHub Check: Rust packages (dashpay-contract) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Linting
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Unused dependencies
- GitHub Check: Rust packages (dashpay-contract) / Formatting
- GitHub Check: Rust packages (dashpay-contract) / Linting
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (drive) / Tests
- GitHub Check: Rust packages (drive) / Linting
- GitHub Check: Rust packages (dpp) / Formatting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Linting
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Linting
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Unused dependencies
- GitHub Check: Rust packages (dashpay-contract) / Formatting
- GitHub Check: Rust packages (dashpay-contract) / Linting
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (drive) / Tests
- GitHub Check: Rust packages (drive) / Linting
- GitHub Check: Rust packages (dpp) / Formatting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Linting
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Linting
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Unused dependencies
- GitHub Check: Rust packages (dashpay-contract) / Formatting
- GitHub Check: Rust packages (dashpay-contract) / Linting
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (drive) / Tests
- GitHub Check: Rust packages (drive) / Linting
- GitHub Check: Rust packages (dpp) / Formatting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Linting
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Linting
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Unused dependencies
- GitHub Check: Rust packages (dashpay-contract) / Formatting
- GitHub Check: Rust packages (dashpay-contract) / Linting
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (drive) / Tests
- GitHub Check: Rust packages (drive) / Linting
- GitHub Check: Rust packages (dpp) / Formatting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Linting
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Unused dependencies
- GitHub Check: Rust packages (dashpay-contract) / Formatting
- GitHub Check: Rust packages (dashpay-contract) / Linting
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (drive) / Tests
- GitHub Check: Rust packages (drive) / Linting
- GitHub Check: Rust packages (dpp) / Formatting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Linting
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Unused dependencies
- GitHub Check: Rust packages (dashpay-contract) / Formatting
- GitHub Check: Rust packages (dashpay-contract) / Linting
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (drive) / Tests
- GitHub Check: Rust packages (drive) / Linting
- GitHub Check: Rust packages (dpp) / Formatting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Linting
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Unused dependencies
- GitHub Check: Rust packages (dashpay-contract) / Formatting
- GitHub Check: Rust packages (dashpay-contract) / Linting
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (drive) / Tests
- GitHub Check: Rust packages (drive) / Linting
- GitHub Check: Rust packages (dpp) / Formatting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (dashpay-contract) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Linting
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Unused dependencies
- GitHub Check: Rust packages (dashpay-contract) / Formatting
- GitHub Check: Rust packages (dashpay-contract) / Linting
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (drive) / Tests
- GitHub Check: Rust packages (drive) / Linting
- GitHub Check: Rust packages (dpp) / Formatting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (dashpay-contract) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Linting
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Unused dependencies
- GitHub Check: Rust packages (dashpay-contract) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (drive) / Tests
- GitHub Check: Rust packages (drive) / Linting
- GitHub Check: Rust packages (dpp) / Formatting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (dashpay-contract) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Linting
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Tests
- GitHub Check: Rust packages (dashpay-contract) / Unused dependencies
- GitHub Check: Rust packages (dashpay-contract) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (drive) / Tests
- GitHub Check: Rust packages (dpp) / Formatting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (dpp) / Tests
🔇 Additional comments (1)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (1)
142-177: Subscription rollback on partial failures — nice defensive fix.Tracks successes and unsubscribes on error; prevents partial subscriptions.
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (5)
packages/rs-dapi/src/services/platform_service/get_status.rs (1)
188-194: Log parse failures for protocol version fields.Parse failures are silently ignored with
unwrap_or(0), which can mask data quality issues from Tenderdash responses.Apply this diff to add debug logging for parse failures:
if !protocol_version.block.is_empty() { - tenderdash_protocol.block = protocol_version.block.parse().unwrap_or(0); + tenderdash_protocol.block = protocol_version.block.parse().unwrap_or_else(|e| { + debug!(error = ?e, value = %protocol_version.block, "Failed to parse Tenderdash block version"); + 0 + }); } if !protocol_version.p2p.is_empty() { - tenderdash_protocol.p2p = protocol_version.p2p.parse().unwrap_or(0); + tenderdash_protocol.p2p = protocol_version.p2p.parse().unwrap_or_else(|e| { + debug!(error = ?e, value = %protocol_version.p2p, "Failed to parse Tenderdash p2p version"); + 0 + }); }packages/rs-dapi/src/services/streaming_service/block_header_stream.rs (1)
339-359: Fix dedupe: use HashSet::insert to detect duplicates (avoid remove+reinsert).
Current remove()+insert removes the seen hash, so the same block may be forwarded again later. Detect duplicates via insert; if insert returns false, skip immediately.Apply this diff:
- let mut allow_forward = true; - - { - // scope for the lock - let mut hashes = delivered_hashes.lock().await; - if hashes.remove(&hash_bytes[..]) { - trace!( - subscriber_id, - block_hash = %block_hash_hex, - "block_headers=skip_duplicate_block" - ); - allow_forward = false; - } else { - hashes.insert(hash_bytes.into()); - } - } - - if !allow_forward { - return true; - } + { + let mut hashes = delivered_hashes.lock().await; + // Insert returns false if already present → duplicate + if !hashes.insert(hash_bytes.to_vec()) { + trace!( + subscriber_id, + block_hash = %block_hash_hex, + "block_headers=skip_duplicate_block" + ); + return true; + } + }packages/rs-dapi/src/services/streaming_service/mod.rs (1)
326-328: Log the actual skipped count in structured field (not a boolean)- Err(RecvError::Lagged(skipped)) => { - tracing::error!(skipped, "ZMQ event reader lagged, skipped events"); - } + Err(RecvError::Lagged(skipped)) => { + debug!(skipped = skipped, "ZMQ receiver lagged; skipping"); + }packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (1)
941-955: Remove raw-bytes bloom fallback and unwrap; treat decode errors as non-matchFalling back to
guard.contains(tx_bytes)misrepresents bloom semantics and masks encoding issues;.unwrap()can panic. Return non‑match on decode failure and drop the lock unwrap.Apply:
- match deserialize::<CoreTx>(tx_bytes.as_slice()) { - Ok(tx) => { - let matches = super::bloom::matches_transaction( - Arc::clone(bloom), - &tx, - *flags, - ); - trace!(height,matches, txid = %tx.txid(), "transactions_with_proofs=bloom_match"); - matches - } - Err(e) => { - debug!(height, error = %e, "transactions_with_proofs=tx_deserialize_failed, checking raw-bytes contains()"); - let guard = bloom.read().unwrap(); - guard.contains(tx_bytes) - } - } + match deserialize::<CoreTx>(tx_bytes.as_slice()) { + Ok(tx) => { + let matches = super::bloom::matches_transaction( + Arc::clone(bloom), + &tx, + *flags, + ); + trace!(height, matches, txid = %tx.txid(), "transactions_with_proofs=bloom_match"); + matches + } + Err(e) => { + debug!(height, error = %e, "transactions_with_proofs=tx_deserialize_failed; treating as non-match"); + false + } + }Based on learnings.
packages/rs-dapi/src/sync.rs (1)
7-10: Return a real abort handle and make tasks abortable
JoinSet::spawndoes not yield atokio::task::AbortHandle. Usefutures::future::abortableand return the futuresAbortHandlevia the oneshot. MapAbortedtoOk(())so cancellations aren’t logged as errors.@@ -use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}; -use tokio::sync::{Mutex, Notify, OnceCell, oneshot}; -use tokio::task::{AbortHandle, JoinError, JoinSet}; +use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}; +use tokio::sync::{Mutex, Notify, OnceCell, oneshot}; +use tokio::task::{JoinError, JoinSet}; +use futures::future::{abortable, AbortHandle as FutAbortHandle}; @@ enum WorkerCommand { Spawn { task: WorkerTask, - response: oneshot::Sender<AbortHandle>, + response: oneshot::Sender<FutAbortHandle>, }, } @@ struct WorkerTaskHandleInner { - handle: OnceCell<Result<AbortHandle, ()>>, - receiver: Mutex<Option<oneshot::Receiver<AbortHandle>>>, + handle: OnceCell<Result<FutAbortHandle, ()>>, + receiver: Mutex<Option<oneshot::Receiver<FutAbortHandle>>>, notify: Notify, } @@ - fn new(receiver: oneshot::Receiver<AbortHandle>) -> Self { + fn new(receiver: oneshot::Receiver<FutAbortHandle>) -> Self { @@ - async fn get_handle(&self) -> Option<AbortHandle> { + async fn get_handle(&self) -> Option<FutAbortHandle> { @@ - async fn take_receiver(&self) -> Option<oneshot::Receiver<AbortHandle>> { + async fn take_receiver(&self) -> Option<oneshot::Receiver<FutAbortHandle>> { @@ - Some(WorkerCommand::Spawn { task, response }) => { - let abort_handle = join_set.spawn(task); - let _ = response.send(abort_handle); - } + Some(WorkerCommand::Spawn { task, response }) => { + let (ab_fut, abort_handle) = abortable(task); + // Map Aborted -> Ok(()) so cancels are clean + join_set.spawn(async move { + match ab_fut.await { + Ok(res) => res, + Err(_aborted) => Ok(()), + } + }); + let _ = response.send(abort_handle); + } @@ - Some(WorkerCommand::Spawn { task, response }) => { - let abort_handle = join_set.spawn(task); - let _ = response.send(abort_handle); - } + Some(WorkerCommand::Spawn { task, response }) => { + let (ab_fut, abort_handle) = abortable(task); + join_set.spawn(async move { + match ab_fut.await { + Ok(res) => res, + Err(_aborted) => Ok(()), + } + }); + let _ = response.send(abort_handle); + }Based on learnings.
Also applies to: 56-61, 122-146, 155-159, 161-186, 200-218, 223-237
🧹 Nitpick comments (11)
packages/rs-dapi/src/server/metrics.rs (2)
66-138: Consider extracting common error-handling pattern.Both platform and core health checks follow the same
Ok(Ok(...)),Ok(Err(...)),Err(timeout)pattern with similar response construction. A helper function could reduce duplication and improve maintainability.Example:
fn map_health_result<T, F>( result: Result<Result<T, DapiError>, tokio::time::error::Elapsed>, on_success: F, ) -> (bool, impl Serialize) where F: FnOnce(T) -> (bool, impl Serialize), { match result { Ok(Ok(data)) => on_success(data), Ok(Err(err)) => { error!(error = %err, "Health check failed"); (false, /* error payload with redacted error */) } Err(_) => (false, /* timeout payload */), } }
170-178: Log error if response building fails.The
unwrap_or_elseon line 177 silently returns an empty response if building the metrics response fails. Consider logging the error to aid debugging..body(axum::body::Body::from(body)) - .unwrap_or_else(|_| axum::response::Response::new(axum::body::Body::from(""))) + .unwrap_or_else(|err| { + error!("Failed to build metrics response: {}", err); + axum::response::Response::new(axum::body::Body::from("")) + })packages/rs-dapi/src/services/platform_service/get_status.rs (3)
205-206: Consider logging saturation when clamping protocol versions.The
min(u32::MAX as u64)approach silently saturates values that exceedu32::MAX. While overflow is unlikely in practice, logging would improve observability.Apply this diff to add warnings:
let drive_protocol_version = get_status_response_v0::version::protocol::Drive { - current: drive_protocol.current.unwrap_or(0).min(u32::MAX as u64) as u32, - latest: drive_protocol.latest.unwrap_or(0).min(u32::MAX as u64) as u32, + current: { + let val = drive_protocol.current.unwrap_or(0); + if val > u32::MAX as u64 { + debug!(value = val, "Drive protocol current version exceeds u32::MAX, saturating"); + } + val.min(u32::MAX as u64) as u32 + }, + latest: { + let val = drive_protocol.latest.unwrap_or(0); + if val > u32::MAX as u64 { + debug!(value = val, "Drive protocol latest version exceeds u32::MAX, saturating"); + } + val.min(u32::MAX as u64) as u32 + }, };
365-366: Consider logging saturation for total_snapshots.Similar to protocol versions, the
min(u32::MAX as u64)saturation is silent. Adding a debug log would improve observability for the unlikely overflow case.Apply this diff:
- total_snapshots: parse_or_default(&sync_info.total_snapshots).min(u32::MAX as u64) - as u32, + total_snapshots: { + let val = parse_or_default(&sync_info.total_snapshots); + if val > u32::MAX as u64 { + debug!(value = val, "total_snapshots exceeds u32::MAX, saturating"); + } + val.min(u32::MAX as u64) as u32 + },
407-407: Consider logging saturation for epoch.The epoch field uses silent saturation. Adding a debug log would improve observability.
Apply this diff:
- time.epoch = drive_time.epoch.map(|e| e.min(u32::MAX as u64) as u32); + time.epoch = drive_time.epoch.map(|e| { + if e > u32::MAX as u64 { + debug!(value = e, "epoch exceeds u32::MAX, saturating"); + } + e.min(u32::MAX as u64) as u32 + });packages/rs-dapi/src/services/streaming_service/block_header_stream.rs (3)
360-370: Use a named constant for header length and enforce exact 80-byte headers in historical path.
Improves clarity and catches malformed headers precisely.Apply this diff:
@@ -const MAX_HEADERS_PER_BATCH: usize = 500; +const MAX_HEADERS_PER_BATCH: usize = 500; +const BLOCK_HEADER_LEN: usize = 80; @@ - if data.len() < 80 { + if data.len() < BLOCK_HEADER_LEN { @@ - let header_bytes = data[..80].to_vec(); + let header_bytes = data[..BLOCK_HEADER_LEN].to_vec(); @@ - if header_bytes.len() < 80 { - return Err(Status::internal( - "Received malformed block header (len < 80)", - )); - } - - response_headers.push(header_bytes[..80].to_vec()); + if header_bytes.len() != BLOCK_HEADER_LEN { + return Err(Status::internal("Received malformed block header (len != 80)")); + } + response_headers.push(header_bytes);Also applies to: 545-552
422-433: Simplify send path; avoid let-chains with await and cloning the response.
Improves readability and avoids cloning large payloads for logging.Apply this diff:
- if let Some(response) = maybe_response.clone() - && tx.send(response).await.is_err() - { - debug!(subscriber_id, "block_headers=client_disconnected"); - return false; - } - - trace!( - subscriber_id, - response=?maybe_response, "block_headers=event_forwarded" - ); + if let Some(response) = maybe_response { + if tx.send(response).await.is_err() { + debug!(subscriber_id, "block_headers=client_disconnected"); + return false; + } + trace!(subscriber_id, "block_headers=event_forwarded"); + }
331-337: Reduce log payload on invalid block parsing.
Logging the entire block hex is heavy and noisy. Log size only (or a short prefix) instead.Apply this diff:
- warn!( - subscriber_id, - block = %hex::encode(&data), - "block_headers=forward_block_invalid_block - it should not happen, report this issue" - ); + warn!( + subscriber_id, + payload_size = data.len(), + "block_headers=forward_block_invalid_block" + );packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (2)
488-501: Avoid cloning large responses just for logging
resp.clone()duplicates potentially large payloads. Log a concise summary before sending, or log fields without?resp, then send the ownedresp.- if tx_sender.send(Ok(resp.clone())).await.is_err() { + // Optional: pre-log a compact summary here if needed + if tx_sender.send(Ok(resp)).await.is_err() { debug!( subscriber_id, "transactions_with_proofs=client_disconnected" ); return false; - } else { - trace!( - event = ?resp, - subscriber_id, - handle_id, - "transactions_with_proofs=forward_transaction_event_success" - ); - } + }
44-49: Bound delivered sets to prevent unbounded memory growth on long‑lived streams
delivered_txs/blocks/instant_locksare unbounded HashSets. For long sessions this can grow without limit. Consider LRU, time‑based purge, or per‑height windowing.Also applies to: 102-147
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (1)
597-622: Replace “ping” subscribe/unsubscribe health check with a more reliable signalPeriodically subscribing to a fake topic is indirect and can create socket churn. Track time since last message or use the monitor/connected flag and a timeout to trigger reconnect; optionally add a lightweight Core RPC ping.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
packages/rs-dapi/src/server/metrics.rs(1 hunks)packages/rs-dapi/src/services/platform_service/get_status.rs(1 hunks)packages/rs-dapi/src/services/streaming_service/block_header_stream.rs(1 hunks)packages/rs-dapi/src/services/streaming_service/mod.rs(1 hunks)packages/rs-dapi/src/services/streaming_service/transaction_stream.rs(1 hunks)packages/rs-dapi/src/services/streaming_service/zmq_listener.rs(1 hunks)packages/rs-dapi/src/sync.rs(1 hunks)
🧰 Additional context used
📓 Path-based instructions (2)
**/*.rs
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.rs: Format Rust code with cargo fmt
Run Clippy linter for Rust code
Files:
packages/rs-dapi/src/sync.rspackages/rs-dapi/src/services/platform_service/get_status.rspackages/rs-dapi/src/services/streaming_service/transaction_stream.rspackages/rs-dapi/src/services/streaming_service/mod.rspackages/rs-dapi/src/services/streaming_service/zmq_listener.rspackages/rs-dapi/src/services/streaming_service/block_header_stream.rspackages/rs-dapi/src/server/metrics.rs
packages/rs-*/**/*.rs
📄 CodeRabbit inference engine (AGENTS.md)
packages/rs-*/**/*.rs: Use rustfmt defaults for Rust code (format via cargo fmt --all)
Keep Rust code clippy-clean (cargo clippy)
Name Rust modules in snake_case
Use PascalCase for Rust types
Use SCREAMING_SNAKE_CASE for Rust constants
Files:
packages/rs-dapi/src/sync.rspackages/rs-dapi/src/services/platform_service/get_status.rspackages/rs-dapi/src/services/streaming_service/transaction_stream.rspackages/rs-dapi/src/services/streaming_service/mod.rspackages/rs-dapi/src/services/streaming_service/zmq_listener.rspackages/rs-dapi/src/services/streaming_service/block_header_stream.rspackages/rs-dapi/src/server/metrics.rs
🧠 Learnings (6)
📓 Common learnings
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/dashmate/src/test/constants/services.js:4-4
Timestamp: 2025-10-15T14:45:30.856Z
Learning: In the dashmate codebase (packages/dashmate), during the DAPI Rust migration (rs-dapi), the old service keys `dapi_api` and `dapi_core_streams` are intentionally kept in `generateEnvsFactory.js` for backward compatibility even though the test constants in `src/test/constants/services.js` have been updated to use `rs_dapi`. These deprecated keys will be removed in a future PR after the transition is complete.
Learnt from: shumkov
PR: dashpay/platform#2270
File: packages/dapi/lib/grpcServer/handlers/platform/broadcastStateTransitionHandlerFactory.js:69-125
Timestamp: 2024-10-24T05:07:35.892Z
Learning: Future development plans involve rewriting DAPI into Rust, as the rest of the project has already migrated.
📚 Learning: 2025-10-09T15:59:33.162Z
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/rs-dapi/src/clients/tenderdash_client.rs:0-0
Timestamp: 2025-10-09T15:59:33.162Z
Learning: In the rs-dapi codebase, the `Workers::spawn` method (in `packages/rs-dapi/src/sync.rs`) accepts futures that return `Result<O, E>`, not `()`. Its signature is `fn spawn<F, O, E>(&self, fut: F) where F: Future<Output = Result<O, E>> + Send + 'static`. The method handles error logging internally, so spawned tasks should return Results directly without additional error handling wrappers.
Applied to files:
packages/rs-dapi/src/sync.rs
📚 Learning: 2025-10-09T15:59:41.943Z
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/rs-dapi/src/clients/tenderdash_client.rs:0-0
Timestamp: 2025-10-09T15:59:41.943Z
Learning: In packages/rs-dapi, the Workers::spawn method accepts futures that return Result<O, E> where E: Into<DapiError>. It handles error logging internally, so spawned tasks should return Result types rather than handling errors inside the task. The Workers implementation automatically logs worker failures.
Applied to files:
packages/rs-dapi/src/sync.rs
📚 Learning: 2024-10-10T10:30:24.653Z
Learnt from: lklimek
PR: dashpay/platform#2232
File: packages/rs-sdk/src/sync.rs:68-71
Timestamp: 2024-10-10T10:30:24.653Z
Learning: In synchronous code within a Tokio runtime, we cannot await spawned task handles (`JoinHandle`), so it's acceptable to check if the task is finished and abort it if necessary.
Applied to files:
packages/rs-dapi/src/sync.rs
📚 Learning: 2025-04-21T00:38:44.796Z
Learnt from: QuantumExplorer
PR: dashpay/platform#2559
File: packages/rs-drive-abci/src/query/token_queries/token_perpetual_distribution_last_claim/v0/mod.rs:128-0
Timestamp: 2025-04-21T00:38:44.796Z
Learning: In the Dash Platform codebase, EpochIndex is defined as u16, making it safe to cast to u32 without overflow concerns.
Applied to files:
packages/rs-dapi/src/services/platform_service/get_status.rs
📚 Learning: 2025-02-03T23:39:10.579Z
Learnt from: QuantumExplorer
PR: dashpay/platform#2450
File: packages/rs-dpp/src/data_contract/associated_token/token_perpetual_distribution/v0/methods.rs:10-12
Timestamp: 2025-02-03T23:39:10.579Z
Learning: Block interval calculations in token distribution logic should use checked arithmetic operations (checked_sub, checked_add) to prevent potential overflows, especially when dealing with block heights and intervals.
Applied to files:
packages/rs-dapi/src/services/streaming_service/block_header_stream.rs
🧬 Code graph analysis (7)
packages/rs-dapi/src/sync.rs (5)
packages/rs-dapi/src/cache.rs (4)
std(125-125)new(58-60)new(96-109)new(148-171)packages/rs-dapi/src/services/streaming_service/mod.rs (1)
new(48-67)packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (5)
new(128-171)new(332-344)spawn(558-561)default(60-72)recv(304-317)packages/rs-dapi/src/services/platform_service/mod.rs (1)
new(110-163)packages/rs-dapi/src/services/streaming_service/masternode_list_sync.rs (1)
new(32-41)
packages/rs-dapi/src/services/platform_service/get_status.rs (4)
packages/rs-dapi/src/cache.rs (13)
cache(474-475)cache(484-485)cache(494-495)cache(518-519)cache(536-537)cache(542-543)cache(577-578)cache(603-604)cache(623-624)make_cache_key(388-401)new(58-60)new(96-109)new(148-171)packages/rs-dapi/src/services/platform_service/mod.rs (1)
new(110-163)packages/rs-dapi/src/clients/drive_client.rs (2)
new(95-128)get_drive_status(167-227)packages/rs-dapi/src/clients/tenderdash_client.rs (4)
new(53-60)new(383-422)status(450-455)net_info(458-472)
packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (2)
packages/rs-dapi/src/services/streaming_service/bloom.rs (1)
bloom_flags_from_int(103-114)packages/rs-dapi/src/services/streaming_service/mod.rs (8)
txid_bytes_from_bytes(355-363)txid_hex_from_bytes(346-352)block_hash_from_block_bytes(365-377)summarize_streaming_event(396-445)deserialize(349-349)deserialize(360-360)deserialize(368-368)short_hex(386-393)
packages/rs-dapi/src/services/streaming_service/mod.rs (4)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (2)
new(128-171)new(332-344)packages/rs-dapi/src/sync.rs (3)
new(23-27)new(74-81)new(136-145)packages/rs-dapi/src/services/streaming_service/masternode_list_sync.rs (1)
new(32-41)packages/rs-dapi/src/clients/tenderdash_websocket.rs (1)
new(78-88)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (2)
packages/rs-dapi/src/sync.rs (6)
default(117-119)new(23-27)new(74-81)new(136-145)spawn(84-112)spawn(202-206)packages/rs-dapi/src/services/streaming_service/mod.rs (2)
new(48-67)summarize_zmq_event(448-489)
packages/rs-dapi/src/services/streaming_service/block_header_stream.rs (2)
packages/rs-dapi/src/services/streaming_service/mod.rs (5)
deserialize(349-349)deserialize(360-360)deserialize(368-368)block_hash_from_block_bytes(365-377)summarize_streaming_event(396-445)packages/rs-dash-event-bus/src/event_bus.rs (2)
id(234-236)recv(239-242)
packages/rs-dapi/src/server/metrics.rs (3)
packages/rs-dapi/src/logging/middleware.rs (1)
new(27-29)packages/rs-dapi/src/metrics.rs (1)
gather_prometheus(383-392)packages/rs-dapi/src/services/platform_service/mod.rs (1)
new(110-163)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (40)
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
🔇 Additional comments (9)
packages/rs-dapi/src/server/metrics.rs (4)
1-43: LGTM: Clean server startup logic.The metrics server initialization is well-structured with appropriate early returns for disabled metrics, conditional access logging, and proper error propagation.
180-227: LGTM: Well-structured response types.The serialization structures are properly designed with appropriate use of
Option,skip_serializing_if, andserde(rename)for consistent JSON field naming.
229-262: LGTM: Comprehensive error redaction.The
health_error_labelfunction properly redacts error details for public health endpoints, preventing information leakage while maintaining useful status information. The catch-all pattern ensures safety.
279-286: ****The review comment is based on incorrect assumptions about field types. The struct
PlatformStatusHealth(defined inpackages/rs-dapi/src/services/platform_service/get_status.rslines 20-24) has error fields typed asOption<DapiError>, notOption<String>. The conversion at metrics.rs lines 76-79 correctly chains throughOption<&DapiError>→From<&DapiError>, which applieshealth_error_labelredaction. TheFrom<String>implementation (lines 279-286) is not used in this code path.Likely an incorrect or invalid review comment.
packages/rs-dapi/src/services/platform_service/get_status.rs (3)
18-89: LGTM: Well-designed health tracking and caching.The
PlatformStatusHealthstruct provides clear component-level health predicates, and theget_status_implhandler appropriately uses a 10-second TTL cache to avoid overwhelming upstream services. Timestamp handling at line 69 correctly clamps to non-negative values before casting.
92-144: LGTM: Safe concurrent upstream fetches with proper error handling.The use of
tokio::join!for parallel fetches (Drive status, Tenderdash status, netinfo) is efficient and safe. Each component failure is logged at debug level and tracked inPlatformStatusHealth, with sensible defaults applied when upstream calls fail.
415-469: LGTM: Comprehensive test validates response construction.The test
build_status_response_populates_fields_from_tenderdash_statusvalidates key fields across version, protocol, node, chain, network, state_sync, and time sections using realistic Tenderdash JSON payload.packages/rs-dapi/src/services/streaming_service/block_header_stream.rs (2)
44-63: Param validation looks good (distinct errors, early rejects).
Clear separation of zero height, empty hash, and missing from_block. Nice.
463-468: Nice use of checked arithmetic for 'available'.
Using checked_sub + checked_add prevents underflow/overflow on height math.Based on learnings
Also applies to: 478-485
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (4)
packages/rs-dapi/src/clients/tenderdash_client.rs (1)
511-518: Document the fallback behavior for subscription methods.Both subscription methods return a broadcast receiver from the WebSocket client. When WebSocket is unavailable, these receivers will never produce events. This fallback behavior should be documented to guide callers.
Add doc comments to clarify the behavior:
+ /// Subscribe to streaming Tenderdash transaction events via WebSocket. + /// + /// Returns a broadcast receiver for transaction events. The receiver will produce events + /// only while the WebSocket connection is active. Check `is_websocket_connected()` + /// to verify connectivity before subscribing, or handle the case where no events arrive. pub fn subscribe_to_transactions(&self) -> broadcast::Receiver<TransactionEvent> { self.websocket_client.subscribe() } + + /// Subscribe to block events from Tenderdash via WebSocket. + /// + /// Returns a broadcast receiver for block events. The receiver will produce events + /// only while the WebSocket connection is active. Check `is_websocket_connected()` + /// to verify connectivity before subscribing, or handle the case where no events arrive. pub fn subscribe_to_blocks(&self) -> broadcast::Receiver<BlockEvent> { self.websocket_client.subscribe_blocks() }packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (3)
433-441: InstantLock filter should default to non‑match when tx bytes are missingThis is the same issue flagged in the previous review. The code still uses
unwrap_or(true)which forwards InstantLocks even when the transaction cannot be decoded. This should default tofalseunlessCoreAllTxsor already delivered.Apply this diff:
FilterType::CoreBloomFilter(bloom, flags) => tx .as_ref() .map(|tx| super::bloom::matches_transaction(Arc::clone(bloom), tx, *flags)) - .unwrap_or(true), + .unwrap_or(false),
951-956: Do not fallback to raw-bytescontains()when tx deserialization failsThis is the same issue flagged in the previous review (lines 937-955 in the old code, marked as addressed in commits 8024cc9 to cd45073). However, the code still contains the raw-bytes fallback pattern that misrepresents bloom semantics and hides encoding issues. Treat deserialization failures as non-match instead.
Apply this diff:
Err(e) => { - debug!(height, error = %e, "transactions_with_proofs=tx_deserialize_failed, checking raw-bytes contains()"); - let guard = bloom.read().unwrap(); - guard.contains(tx_bytes) + debug!(height, error = %e, "transactions_with_proofs=tx_deserialize_failed; treating as non-match"); + false }
953-953: Avoid.unwrap()on lock acquisitionThis is the same issue flagged in the previous review (marked as addressed in commit 656763f). The
.unwrap()will panic if the RwLock is poisoned. Use proper error handling orexpectwith a descriptive message. This line is part of the fallback pattern that should be removed entirely per the previous comment.If the raw-bytes fallback is removed as suggested above, this line will be removed automatically. Otherwise, replace with proper error handling:
- let guard = bloom.read().unwrap(); + let guard = bloom.read().inspect_err(|e| { + debug!(height, error = ?e, "transactions_with_proofs=bloom_lock_poisoned"); + })?;
🧹 Nitpick comments (4)
packages/rs-dapi/src/clients/tenderdash_client.rs (1)
328-334: Remove redundant Content-Type header.The
.json()method automatically sets theContent-Type: application/jsonheader, making line 331 redundant.Apply this diff:
let response_body = self .client .post(&self.base_url) - .header("Content-Type", "application/json") .json(request) .send() .awaitpackages/rs-dapi/src/server/metrics.rs (2)
38-39: Consider adding error context for bind/serve failures.The bind and serve operations propagate raw IO errors. Adding context would help operators diagnose configuration or port-binding issues more quickly.
Apply this diff to add error context:
- let listener = TcpListener::bind(addr).await?; - axum::serve(listener, app).await?; + let listener = TcpListener::bind(addr).await + .map_err(|e| DapiError::Io(format!("Failed to bind metrics server to {}: {}", addr, e)))?; + axum::serve(listener, app).await + .map_err(|e| DapiError::Io(format!("Failed to serve metrics server: {}", e)))?;
170-178: Consider logging response builder errors.The
unwrap_or_elseon line 177 silently swallows response-building errors. While this error is unlikely, logging it would aid debugging.Apply this diff:
.body(axum::body::Body::from(body)) - .unwrap_or_else(|_| axum::response::Response::new(axum::body::Body::from(""))) + .unwrap_or_else(|e| { + error!(error = %e, "Failed to build metrics response"); + axum::response::Response::new(axum::body::Body::from("")) + })packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (1)
1083-1092: Consider avoiding unnecessary clone of bloom_filterThe
bloom_filterparameter is cloned multiple times (line 1083, then its fields are cloned again at 1086-1088). Since this function only reads the fields, you could work with references throughout to reduce allocations.Example refactor (adjust types as needed):
- let bloom_filter_clone = bloom_filter.clone(); - let flags = bloom_flags_from_int(bloom_filter_clone.n_flags); + let flags = bloom_flags_from_int(bloom_filter.n_flags); let core_filter = dashcore_rpc::dashcore::bloom::BloomFilter::from_bytes( - bloom_filter_clone.v_data.clone(), - bloom_filter_clone.n_hash_funcs, - bloom_filter_clone.n_tweak, + bloom_filter.v_data.clone(), + bloom_filter.n_hash_funcs, + bloom_filter.n_tweak, flags, )
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
packages/rs-dapi/src/clients/tenderdash_client.rs(1 hunks)packages/rs-dapi/src/server/metrics.rs(1 hunks)packages/rs-dapi/src/services/streaming_service/transaction_stream.rs(1 hunks)
🧰 Additional context used
📓 Path-based instructions (2)
**/*.rs
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.rs: Format Rust code with cargo fmt
Run Clippy linter for Rust code
Files:
packages/rs-dapi/src/services/streaming_service/transaction_stream.rspackages/rs-dapi/src/server/metrics.rspackages/rs-dapi/src/clients/tenderdash_client.rs
packages/rs-*/**/*.rs
📄 CodeRabbit inference engine (AGENTS.md)
packages/rs-*/**/*.rs: Use rustfmt defaults for Rust code (format via cargo fmt --all)
Keep Rust code clippy-clean (cargo clippy)
Name Rust modules in snake_case
Use PascalCase for Rust types
Use SCREAMING_SNAKE_CASE for Rust constants
Files:
packages/rs-dapi/src/services/streaming_service/transaction_stream.rspackages/rs-dapi/src/server/metrics.rspackages/rs-dapi/src/clients/tenderdash_client.rs
🧠 Learnings (3)
📓 Common learnings
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/dashmate/src/test/constants/services.js:4-4
Timestamp: 2025-10-15T14:45:30.856Z
Learning: In the dashmate codebase (packages/dashmate), during the DAPI Rust migration (rs-dapi), the old service keys `dapi_api` and `dapi_core_streams` are intentionally kept in `generateEnvsFactory.js` for backward compatibility even though the test constants in `src/test/constants/services.js` have been updated to use `rs_dapi`. These deprecated keys will be removed in a future PR after the transition is complete.
📚 Learning: 2025-10-09T15:59:33.162Z
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/rs-dapi/src/clients/tenderdash_client.rs:0-0
Timestamp: 2025-10-09T15:59:33.162Z
Learning: In the rs-dapi codebase, the `Workers::spawn` method (in `packages/rs-dapi/src/sync.rs`) accepts futures that return `Result<O, E>`, not `()`. Its signature is `fn spawn<F, O, E>(&self, fut: F) where F: Future<Output = Result<O, E>> + Send + 'static`. The method handles error logging internally, so spawned tasks should return Results directly without additional error handling wrappers.
Applied to files:
packages/rs-dapi/src/clients/tenderdash_client.rs
📚 Learning: 2025-10-09T15:59:41.943Z
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/rs-dapi/src/clients/tenderdash_client.rs:0-0
Timestamp: 2025-10-09T15:59:41.943Z
Learning: In packages/rs-dapi, the Workers::spawn method accepts futures that return Result<O, E> where E: Into<DapiError>. It handles error logging internally, so spawned tasks should return Result types rather than handling errors inside the task. The Workers implementation automatically logs worker failures.
Applied to files:
packages/rs-dapi/src/clients/tenderdash_client.rs
🧬 Code graph analysis (3)
packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (2)
packages/rs-dapi/src/services/streaming_service/bloom.rs (1)
bloom_flags_from_int(103-114)packages/rs-dapi/src/services/streaming_service/mod.rs (8)
txid_bytes_from_bytes(355-363)txid_hex_from_bytes(346-352)block_hash_from_block_bytes(365-377)summarize_streaming_event(396-445)deserialize(349-349)deserialize(360-360)deserialize(368-368)short_hex(386-393)
packages/rs-dapi/src/server/metrics.rs (4)
packages/rs-dapi/src/logging/middleware.rs (1)
new(27-29)packages/rs-dapi/src/metrics.rs (1)
gather_prometheus(383-392)packages/rs-dapi/src/clients/tenderdash_client.rs (2)
new(53-60)new(378-417)packages/rs-dapi/src/server/mod.rs (1)
new(29-77)
packages/rs-dapi/src/clients/tenderdash_client.rs (4)
packages/rs-dapi/src/utils.rs (3)
deserialize_string_number_or_null(147-163)deserialize_string_or_number(23-86)generate_jsonrpc_id(12-21)packages/rs-dapi/src/clients/tenderdash_websocket.rs (2)
new(78-88)test_connection(106-125)packages/rs-dapi/src/server/mod.rs (1)
new(29-77)packages/rs-dapi/src/error.rs (3)
from_tenderdash_error(170-172)client(201-203)server_unavailable(231-233)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: Rust crates security audit
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Swift SDK and Example build (warnings as errors)
🔇 Additional comments (12)
packages/rs-dapi/src/clients/tenderdash_client.rs (7)
1-34: LGTM: Well-structured client with clear documentation.The imports are appropriate, and the struct documentation clearly explains the tracing middleware integration and error handling philosophy. The use of
ClientWithMiddlewareandArc<TenderdashWebSocketClient>is correct for shared client patterns.
36-61: LGTM: Clean JSON-RPC request/response types.The generic
TenderdashResponse<T>andJsonRpcRequest<T>types follow standard JSON-RPC 2.0 patterns. The use ofgenerate_jsonrpc_id()ensures unique request IDs.
63-85: LGTM: Efficient parameter structs with appropriate lifetimes.The parameter structs use borrowed
&'a strto avoid unnecessary string copies during RPC calls. TheEmptyParamstype is a clean pattern for parameterless RPCs.
87-301: LGTM: Comprehensive and robust data models.The response models handle Tenderdash's flexible JSON encoding well:
#[serde(default)]handles optional/missing fields gracefully- Custom deserializers (
deserialize_string_or_number,deserialize_string_number_or_null) handle Tenderdash's inconsistent number encoding- Public type aliases improve API ergonomics
303-314: LGTM: Useful helper for detecting empty results.The
is_empty()method provides a clean way to detect default/absent transaction results, which is valuable for downstream error handling.
374-442: LGTM: Robust client construction with proper validation.The constructor properly addresses previous review concerns:
- HTTP client has both
connect_timeoutand requesttimeoutconfigured (lines 385-388)TracingMiddlewareis correctly wired (line 395)- Both HTTP and WebSocket connectivity are validated before returning the client
- Error handling appropriately logs and propagates failures
444-510: LGTM: Clean RPC wrapper methods with consistent patterns.The RPC wrapper methods follow a consistent pattern and properly delegate to the generic
post<T, R>method. Thenet_infomethod's error masking behavior (returning defaults on failure) is intentional and documented in the inline comment.packages/rs-dapi/src/server/metrics.rs (4)
1-11: LGTM!Imports are well-organized and appropriate for the metrics server functionality.
180-227: LGTM!The response payload structs are well-designed with:
- Proper serde attributes for JSON serialization
- Consistent camelCase field naming for external API
- Optional fields correctly marked with
skip_serializing_if
229-266: LGTM!The
health_error_labelfunction correctly implements error redaction:
- Maps all DapiError variants to generic, public-safe labels
- Uses exhaustive matching (no default) to ensure new error types are explicitly handled
- Preserves detailed errors in trace logs for internal debugging
This pattern effectively prevents information leakage while maintaining observability.
45-168: Verification complete—code is secure and no issues found.The verification confirms the error fields in
PlatformStatusHealthareOption<DapiError>(notOption<String>), so the.as_ref().into()conversions at lines 76-79 correctly invoke theFrom<&DapiError>implementation with proper redaction. All component-level errors are safely redacted before exposure in the health response, and the overall health check aggregation logic is sound.packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (1)
1103-1256: Solid test coverage for core streaming functionalityThe test suite covers the critical paths: deduplication, gate synchronization, pending event flushing, merkle block construction, and error handling. Well done on the comprehensive test coverage.
There was a problem hiding this comment.
Actionable comments posted: 4
♻️ Duplicate comments (4)
packages/rs-dapi/src/clients/drive_client.rs (1)
177-177: Critical: Compilation failure from dereferencing non-CopyGetStatusRequest.
GetStatusRequestis a protobuf-generated struct that is notCopy. Line 177 attempts*request, which tries to move the value out of a shared reference and will fail to compile.Apply this diff to clone the request before passing it to
get_status:- let drive_response = self.get_client().get_status(*request).await?.into_inner(); + let drive_response = self + .get_client() + .get_status(request.clone()) + .await? + .into_inner();packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (1)
376-379: Update docstring to match renamed method.The method was correctly renamed from
is_connected()tois_running()per previous review, but the docstring still says "is connected (placeholder)".Apply this diff:
- /// Check if the ZMQ listener is connected (placeholder) + /// Check if the ZMQ listener is still running (not cancelled) pub fn is_running(&self) -> bool { !self.cancel.is_cancelled() }packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (2)
941-955: Do not fallback to raw-bytes bloom.contains() on tx decode error; also avoid.unwrap()On deserialization error the code checks raw bytes against the bloom filter and uses
.unwrap()onRwLock, reintroducing the previously flagged semantics and panic risk.- Err(e) => { - debug!(height, error = %e, "transactions_with_proofs=tx_deserialize_failed, checking raw-bytes contains()"); - let guard = bloom.read().unwrap(); - guard.contains(tx_bytes) - } + Err(e) => { + debug!(height, error = %e, "transactions_with_proofs=tx_deserialize_failed; treating as non-match"); + false + }Please update tests to assert that deserialization failures do NOT yield matches and no lock unwraps occur. Based on learnings.
433-441: InstantLock bloom default should be non‑match whentx_bytesmissing
unwrap_or(true)forwards ILs without a decoded tx. Default tofalseunlessCoreAllTxsor an explicit match.- .unwrap_or(true), + .unwrap_or(false),Update tests accordingly. Based on learnings.
🧹 Nitpick comments (9)
packages/rs-dapi/src/clients/drive_client.rs (1)
248-272: Consider adding unit tests for response mapping logic.The current test only validates client creation. Consider adding tests for:
get_drive_statuswith mock responses- Response mapping from protobuf to
DriveStatusResponse- Error handling for malformed responses
Example test structure:
#[tokio::test] async fn test_get_drive_status_response_mapping() { // Mock or fixture-based test that validates the nested Option // handling and type conversions in lines 180-230 }packages/rs-dapi/src/config/tests.rs (1)
27-40: Consider usingtemp_envor similar for safer test environment manipulation.While the
unsafeblocks are correctly documented and the#[serial]attribute prevents races, modern Rust testing offers safer alternatives like thetemp_envcrate orserial_test's built-in env helpers that provide scoped environment variable management withoutunsafe.Example refactor (would require adding
temp_envdependency):// Instead of manual unsafe set/remove, use: use temp_env; #[test] fn test_example() { temp_env::with_vars( [("DAPI_GRPC_SERVER_PORT", Some("7005"))], || { // test code here } ); // vars are automatically restored }packages/rs-dapi/src/main.rs (1)
264-281: Consider removing hardcoded worker thread count.The runtime is configured with a fixed 4 worker threads (line 267). Tokio's default behavior—using the number of available CPU cores—is generally more appropriate and adapts to the deployment environment.
Unless there's a specific reason for limiting to 4 threads, consider removing the explicit
worker_threads(4)call to allow Tokio to auto-configure based on available hardware.fn main() -> Result<(), ExitCode> { let rt = tokio::runtime::Builder::new_multi_thread() - .worker_threads(4) .enable_all() .build() .expect("Failed to create Tokio runtime");packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (2)
597-622: Consider alternative health check approaches.The current subscribe/unsubscribe to "ping" health check (acknowledged as a "hack" on line 600) has potential issues:
- Repeated
subscribe()calls may not be idempotent in the zeromq library, potentially accumulating internal state- The "ping" topic doesn't exist in Dash Core, so this doesn't verify receipt of actual messages
- Using subscribe success as a proxy for connection health is indirect
While the comment indicates this works around monitor limitations, consider these more robust alternatives:
Option 1: Track time since last message
async fn tick_event_10s(&mut self, last_message_time: Arc<Mutex<Instant>>) { let elapsed = last_message_time.lock().await.elapsed(); if elapsed > Duration::from_secs(60) { debug!("No ZMQ messages received for 60s, connection may be stale"); self.connected.store(false, Ordering::SeqCst); self.cancel.cancel(); } }Option 2: Monitor Dash Core RPC health
Send periodic lightweight RPC calls to Dash Core and cancel on failure, rather than testing the ZMQ socket directly.Option 3: Research zeromq socket health APIs
Check if the zeromq crate exposes native health check or keepalive mechanisms that are more reliable than subscribe/unsubscribe cycling.
660-680: Add test coverage for malformed rawtxlocksig payloads.The existing test validates the happy path where both transaction and lock parse successfully. Consider adding a test for the error case where the transaction decode fails, to verify the behavior matches expectations.
Example test:
#[test] fn split_tx_and_lock_handles_malformed_data() { // Invalid transaction data let malformed_data = vec![0xFF; 100]; let (tx_bytes, lock_bytes) = split_tx_and_lock(malformed_data.clone()); // Verify behavior when transaction decode fails // (Update assertions based on the fix you choose for the malformed data handling) assert!(tx_bytes.is_none(), "should not extract tx from malformed data"); // Current behavior: assert!(lock_bytes.is_some()); // Recommended behavior: assert!(lock_bytes.is_none()); }packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (4)
485-501: Avoid cloning responses just for loggingCloning large payloads before send is wasteful. Log first, then send the owned value.
- Ok(resp) => { - if tx_sender.send(Ok(resp.clone())).await.is_err() { + Ok(resp) => { + trace!( + event = ?resp, + subscriber_id, + handle_id, + "transactions_with_proofs=forward_transaction_event_success" + ); + if tx_sender.send(Ok(resp)).await.is_err() { debug!( subscriber_id, "transactions_with_proofs=client_disconnected" ); return false; - } else { - trace!( - event = ?resp, - subscriber_id, - handle_id, - "transactions_with_proofs=forward_transaction_event_success" - ); - } + } }
84-100: Handle dropped gate sender to avoid potential spinIf
receiver.changed().awaitreturns Err (sender dropped), the wait loop breaks but the gate may remain closed, causing repeated immediate wakeups. Open the gate or return early on drop.- while !*receiver.borrow() { - if receiver.changed().await.is_err() { - break; - } - } + while !*receiver.borrow() { + if receiver.changed().await.is_err() { + debug!("transactions_with_proofs=gate_sender_dropped; opening gate defensively"); + let _ = self.gate_sender.send(true); + break; + } + }
805-851: Mempool scan is fully serial; consider bounded parallelismFetching each tx sequentially increases tail latency on large mempools. Use a bounded concurrency (e.g., FuturesUnordered with a limit) to overlap RPCs.
I can provide a small refactor using
FuturesUnorderedwith a concurrency cap (e.g., 25) if desired.
349-356: Log compact summaries instead of full payloadsCurrent logs may include full gRPC payloads (large byte arrays). Prefer concise counters/sizes to keep logs light.
Example: log counts
raw.transactions.len()orbytes.len()instead of?resp.Also applies to: 862-869, 1007-1013
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
packages/rs-dapi/src/clients/drive_client.rs(1 hunks)packages/rs-dapi/src/config/tests.rs(1 hunks)packages/rs-dapi/src/main.rs(1 hunks)packages/rs-dapi/src/services/streaming_service/transaction_stream.rs(1 hunks)packages/rs-dapi/src/services/streaming_service/zmq_listener.rs(1 hunks)
🧰 Additional context used
📓 Path-based instructions (2)
**/*.rs
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.rs: Format Rust code with cargo fmt
Run Clippy linter for Rust code
Files:
packages/rs-dapi/src/services/streaming_service/transaction_stream.rspackages/rs-dapi/src/main.rspackages/rs-dapi/src/clients/drive_client.rspackages/rs-dapi/src/config/tests.rspackages/rs-dapi/src/services/streaming_service/zmq_listener.rs
packages/rs-*/**/*.rs
📄 CodeRabbit inference engine (AGENTS.md)
packages/rs-*/**/*.rs: Use rustfmt defaults for Rust code (format via cargo fmt --all)
Keep Rust code clippy-clean (cargo clippy)
Name Rust modules in snake_case
Use PascalCase for Rust types
Use SCREAMING_SNAKE_CASE for Rust constants
Files:
packages/rs-dapi/src/services/streaming_service/transaction_stream.rspackages/rs-dapi/src/main.rspackages/rs-dapi/src/clients/drive_client.rspackages/rs-dapi/src/config/tests.rspackages/rs-dapi/src/services/streaming_service/zmq_listener.rs
🧠 Learnings (1)
📓 Common learnings
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/dashmate/src/test/constants/services.js:4-4
Timestamp: 2025-10-15T14:45:30.856Z
Learning: In the dashmate codebase (packages/dashmate), during the DAPI Rust migration (rs-dapi), the old service keys `dapi_api` and `dapi_core_streams` are intentionally kept in `generateEnvsFactory.js` for backward compatibility even though the test constants in `src/test/constants/services.js` have been updated to use `rs_dapi`. These deprecated keys will be removed in a future PR after the transition is complete.
🧬 Code graph analysis (5)
packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (2)
packages/rs-dapi/src/services/streaming_service/bloom.rs (2)
bloom_flags_from_int(103-114)matches_transaction(55-101)packages/rs-dapi/src/services/streaming_service/mod.rs (8)
txid_bytes_from_bytes(355-363)txid_hex_from_bytes(346-352)block_hash_from_block_bytes(365-377)summarize_streaming_event(396-445)deserialize(349-349)deserialize(360-360)deserialize(368-368)short_hex(386-393)
packages/rs-dapi/src/main.rs (5)
packages/rs-dapi/src/config/mod.rs (2)
std(225-225)load_from_dotenv(224-226)packages/rs-dapi/src/logging/mod.rs (1)
init_logging(18-43)packages/rs-dapi/src/error.rs (1)
server(236-238)packages/rs-dapi/src/server/mod.rs (2)
run(82-109)new(29-77)packages/rs-dapi/src/logging/access_log.rs (1)
new(261-272)
packages/rs-dapi/src/clients/drive_client.rs (2)
packages/rs-dapi/src/services/streaming_service/mod.rs (1)
new(48-67)packages/rs-dapi/src/services/platform_service/mod.rs (1)
new(110-163)
packages/rs-dapi/src/config/tests.rs (1)
packages/rs-dapi/src/config/mod.rs (15)
std(225-225)default(47-54)default(136-145)default(149-153)default(157-162)default(166-174)default(198-205)load(210-216)load_from_dotenv(224-226)metrics_port(323-325)load_with_overrides(229-310)grpc_server_addr(313-315)json_rpc_addr(318-320)metrics_addr(333-341)metrics_enabled(328-330)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (3)
packages/rs-dapi/src/sync.rs (6)
default(117-119)new(23-27)new(74-81)new(136-145)spawn(84-112)spawn(202-206)packages/rs-dapi/src/services/streaming_service/mod.rs (2)
new(48-67)summarize_zmq_event(448-489)packages/rs-dapi/src/error.rs (1)
timeout(251-253)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (104)
- GitHub Check: Rust packages (rs-dapi-client) / Check each feature
- GitHub Check: Rust packages (rs-dapi-client) / Detect immutable structure changes
- GitHub Check: Rust packages (rs-dapi-client) / Linting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (rs-dapi-client) / Detect immutable structure changes
- GitHub Check: Rust packages (rs-dapi-client) / Linting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (rs-sdk-ffi) / Unused dependencies
- GitHub Check: Rust packages (rs-dapi-client) / Detect immutable structure changes
- GitHub Check: Rust packages (rs-dapi-client) / Linting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (wasm-dpp) / Formatting
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (rs-sdk-ffi) / Unused dependencies
- GitHub Check: Rust packages (rs-dapi-client) / Linting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (wasm-dpp) / Formatting
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (rs-sdk-ffi) / Unused dependencies
- GitHub Check: Rust packages (wasm-dpp) / Detect immutable structure changes
- GitHub Check: Rust packages (rs-dapi-client) / Linting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (wasm-dpp) / Formatting
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (rs-sdk-ffi) / Unused dependencies
- GitHub Check: Rust packages (wasm-dpp) / Detect immutable structure changes
- GitHub Check: Rust packages (rs-dapi-client) / Linting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (wasm-dpp) / Formatting
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (rs-sdk-ffi) / Unused dependencies
- GitHub Check: Rust packages (wasm-dpp) / Detect immutable structure changes
- GitHub Check: Rust packages (rs-dapi-client) / Linting
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (wasm-dpp) / Formatting
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (rs-sdk-ffi) / Unused dependencies
- GitHub Check: Rust packages (wasm-dpp) / Detect immutable structure changes
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (wasm-dpp) / Formatting
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (rs-sdk-ffi) / Unused dependencies
- GitHub Check: Rust packages (wasm-dpp) / Detect immutable structure changes
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (wasm-dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (wasm-dpp) / Formatting
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (rs-sdk-ffi) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Tests
- GitHub Check: Rust packages (wasm-dpp) / Unused dependencies
- GitHub Check: Rust packages (dpp) / Check each feature
- GitHub Check: Rust packages (wasm-dpp) / Formatting
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
🔇 Additional comments (10)
packages/rs-dapi/src/config/tests.rs (3)
347-404: Socket address tests are comprehensive and well-structured.The tests thoroughly validate:
- Default socket address formatting for gRPC, JSON-RPC, and metrics endpoints
- Custom bind address behavior with non-default ports
- Metrics disabling logic when port is set to 0
The assertions correctly verify both the address resolution and the optional metrics behavior.
406-430: Validation tests correctly verify both success and failure paths.The tests appropriately:
- Confirm default configuration passes validation
- Verify invalid bind addresses are rejected with descriptive error messages
- Include custom assertion messages for better test failure diagnostics
69-86: Asymmetric client connection behavior is confirmed as intentional.The test expectations are correct and match the designed behavior:
DriveClient gracefully degrades: validation failure is logged as a warning, but the client is still returned so the caller can operate in degraded mode. Documentation at lines 92-95 of drive_client.rs explicitly states this design: "If the Drive service cannot be reached, the error is logged and the client is still returned so the caller can operate in a degraded mode while health checks surface the issue."
TenderdashClient requires strict validation: calls both
validate_connection()(HTTP endpoint) andtest_connection()(WebSocket endpoint), returning an error if either fails. This ensures the client cannot operate without server availability.The code reflects this distinction appropriately, and no changes are needed.
packages/rs-dapi/src/main.rs (4)
1-93: Well-structured CLI with comprehensive documentation.The CLI design is excellent:
- Clear subcommand documentation with usage guidance
- Multi-level verbosity support with intuitive semantics
- Auto-detection for color output when not specified
- Debug flag as a convenience shorthand for common development use
The help text appropriately warns about sensitive data in the Config command output.
95-174: Command orchestration is correct and the --force issue has been addressed.The implementation properly:
- Loads configuration before dispatching to command handlers
- Uses
tokio::select!to race server execution against shutdown signals- Pins the server future for polling (line 114) as required by select!
- Maps connection errors to user-friendly messages without mentioning the non-existent --force flag
The previous review's concern about incorrect --force guidance has been resolved.
176-238: Helper functions are well-structured and correct.The helpers appropriately:
- Exit early with error messages on configuration failures
- Map CLI options to logging configuration
- Wrap shared config in
Arcfor the server (line 208)- Warn users about sensitive data when dumping configuration (line 222)
- Provide clean version output
240-262: Cross-platform shutdown signal handling follows best practices.The implementation correctly:
- Handles both SIGTERM and SIGINT on Unix platforms
- Falls back to
tokio::signal::ctrl_c()on non-Unix platforms- Uses conditional compilation to avoid importing Unix-specific APIs on other platforms
This matches the recommended Tokio patterns for graceful shutdown.
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (1)
173-223: Subscription error handling looks solid.The implementation now properly tracks subscribed topics and unsubscribes on failure, addressing the previous review concern. The first-error collection pattern and guaranteed cleanup (even on unsubscribe errors) are both appropriate.
packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (2)
358-376: Block-hash dedupe uses internal byte order (good)Using
block_hash_from_block_bytesand passing its raw hash bytes intomark_block_deliveredaligns with internal representation and fixes prior byte‑order mismatches.
150-167: Raw-bytes bloom fallback found at lines 952–954The verification grep found the pattern you asked to check: lines 952–954 in
transaction_stream.rscontain thetx_deserialize_failederror path that falls back to checking raw bytes against the bloom filter withbloom.read().unwrap()andguard.contains(tx_bytes). This matches the debug log message in your concern.The fallback occurs when transaction deserialization fails and resorts to raw-bytes bloom containment checking as a recovery mechanism. Verify this fallback behavior is intentional and acceptable for the error path, or determine if it requires additional error handling (e.g., handling potential
RwLockpoisoning instead of unwrapping).
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (6)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (2)
520-536: Handle malformed rawtxlocksig payloads defensively.This issue was flagged in a previous review and remains unaddressed. When transaction decoding fails (line 534), returning
(None, Some(data))treats the entire unparseable payload as lock bytes. This is problematic because:
- The rawtxlocksig format is
<transaction><lock_signature>where the transaction comes first- If the transaction can't be decoded, the data format is unexpected and the "lock bytes" position is unknown
- Downstream code in
parse_zmq_message(lines 488-504) receives an event withtx_bytes=Nonebutlock_bytes=<malformed_data>, which may cause issues- No logging occurs when decode fails, making debugging harder
Apply this diff to handle decode errors defensively:
fn split_tx_and_lock(data: Vec<u8>) -> (Option<Vec<u8>>, Option<Vec<u8>>) { let mut cursor = Cursor::new(data.as_slice()); match CoreTransaction::consensus_decode(&mut cursor) { Ok(_) => { let consumed = cursor.position() as usize; if consumed >= data.len() { // Transaction consumed all bytes, no lock data present (Some(data), None) } else { let lock_bytes = data[consumed..].to_vec(); let tx_bytes = data[..consumed].to_vec(); (Some(tx_bytes), Some(lock_bytes)) } } - Err(_) => (None, Some(data)), + Err(e) => { + debug!(error = ?e, data_len = data.len(), "Failed to decode transaction from rawtxlocksig payload"); + (None, None) + } } }Then update
parse_zmq_messageto handle the(None, None)case:let (tx_bytes, lock_bytes_opt) = split_tx_and_lock(data); + if tx_bytes.is_none() && lock_bytes_opt.is_none() { + debug!("rawtxlocksig payload has malformed transaction data, dropping event"); + return None; + } if let Some(lock_bytes) = lock_bytes_opt && !lock_bytes.is_empty() {
604-648: Health check mechanism remains unreliable.This issue was flagged in a previous review and has not been fully addressed. The health check (lines 604-648) has a two-stage approach:
- First checks if a message was received within the last 10 seconds (lines 606-611) ✓
- Falls back to subscribing to a "ping" topic (lines 613-623) ✗
The fallback approach has several problems:
- Subscribing to the same topic repeatedly may not be idempotent—each call could create additional internal state in the ZeroMQ socket
- The "ping" topic is not a real Dash Core topic, so this check doesn't verify that you can actually receive real messages
- Using
subscribe()success/failure as a proxy for connection health is indirect and may not catch all failure modesConsider alternative approaches:
Option 1: Remove the fallback and rely solely on last-message tracking
async fn tick_event_10s(&mut self) { - // first, if we received a message within last 10s, we are connected let last_recv = self.last_recv.load(Ordering::SeqCst); - if last_recv + 10 >= chrono::Utc::now().timestamp() { - self.connected.store(true, Ordering::SeqCst); - return; - } - - // fallback to subscribing to some dummy `ping` topic. - // This is a hack to ensure the connection is alive, as the monitor fails to notify us about disconnects. - let current_status = self.socket.subscribe("ping").await.is_ok(); - // Unsubscribe immediately to avoid resource waste - self.socket - .unsubscribe("ping") - .await - .inspect_err(|e| { - debug!(error = %e, "Error unsubscribing from ping topic during health check"); - }) - .ok(); - - // If the status changed, log it - let previous_status = self.connected.swap(current_status, Ordering::SeqCst); - if current_status != previous_status { - if current_status { - debug!("ZMQ connection recovered"); - } else { - debug!("ZMQ connection is lost, connection will be restarted"); - // disconnect the socket - self.cancel.cancel(); - } - } - - // if we are connected, we assume last_recv is now - if current_status { - self.last_recv - .compare_exchange( - last_recv, - chrono::Utc::now().timestamp(), - Ordering::AcqRel, - Ordering::Relaxed, - ) - .ok(); + let elapsed = chrono::Utc::now().timestamp() - last_recv; + + if elapsed > 60 { // No messages for 60 seconds + debug!("No ZMQ messages received for {} seconds, connection may be dead", elapsed); + self.connected.store(false, Ordering::SeqCst); + self.cancel.cancel(); + } else { + self.connected.store(true, Ordering::SeqCst); } }Option 2: Use Core RPC health checks
Send periodic health check requests to Dash Core via RPC instead of relying on ZMQ subscription operations.packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (2)
433-441: InstantLock bloom filter: default to non‑match when tx bytes are missingForwarding ILs without decoded tx bytes causes false positives. Default to false here.
- FilterType::CoreBloomFilter(bloom, flags) => tx - .as_ref() - .map(|tx| super::bloom::matches_transaction(Arc::clone(bloom), tx, *flags)) - .unwrap_or(true), // failsafe: we assume match to be on a safe side + FilterType::CoreBloomFilter(bloom, flags) => tx + .as_ref() + .map(|tx| super::bloom::matches_transaction(Arc::clone(bloom), tx, *flags)) + .unwrap_or(false),Based on learnings.
941-955: Remove raw‑bytes bloom fallback and unwrap; treat deserialization errors as non‑matchFalling back to
guard.contains(tx_bytes)both misrepresents bloom semantics and reintroduces a panic onRwLock::read().unwrap().- Err(e) => { - debug!(height, error = %e, "transactions_with_proofs=tx_deserialize_failed, checking raw-bytes contains()"); - let guard = bloom.read().unwrap(); - guard.contains(tx_bytes) - } + Err(e) => { + debug!(height, error = %e, "transactions_with_proofs=tx_deserialize_failed; treating as non-match"); + false + }Based on learnings.
packages/rs-dapi/src/services/streaming_service/block_header_stream.rs (1)
339-354: Fix block-hash dedupe: use insert semantics (avoid remove+insert)Using
remove()theninsert()allows later duplicates through. Insert should returnfalsefor duplicates; skip when it does.- { - // scope for the lock - let mut hashes = delivered_hashes.lock().await; - if hashes.remove(&hash_bytes[..]) { - trace!( - subscriber_id, - block_hash = %block_hash_hex, - "block_headers=skip_duplicate_block" - ); - allow_forward = false; - } else { - hashes.insert(hash_bytes.into()); - } - } + { + let mut hashes = delivered_hashes.lock().await; + // Insert returns false if already present → duplicate + if !hashes.insert(hash_bytes.to_vec()) { + trace!( + subscriber_id, + block_hash = %block_hash_hex, + "block_headers=skip_duplicate_block" + ); + allow_forward = false; + } + }Based on learnings.
packages/rs-dapi/src/clients/drive_client.rs (1)
175-178: Compile blocker: don’t move out of&GetStatusRequest; clone before gRPC call
*requestattempts to move a non‑Copy type. Clone the message (prost structs areClone).- let drive_response = self.get_client().get_status(*request).await?.into_inner(); + let drive_response = self + .get_client() + .get_status(request.clone()) + .await? + .into_inner();
🧹 Nitpick comments (6)
packages/rs-dapi/src/config/tests.rs (4)
8-25: Make env cleanup future‑proof (cover all DAPI_ keys).*Enumerating a fixed list risks leaking new keys into tests. Clean all keys with the DAPI_ prefix to ensure isolation.
fn cleanup_env_vars() { - let env_vars = [ - "DAPI_GRPC_SERVER_PORT", - "DAPI_JSON_RPC_PORT", - "DAPI_METRICS_PORT", - "DAPI_BIND_ADDRESS", - "DAPI_DRIVE_URI", - "DAPI_TENDERDASH_URI", - "DAPI_TENDERDASH_WEBSOCKET_URI", - "DAPI_CORE_ZMQ_URL", - "DAPI_STATE_TRANSITION_WAIT_TIMEOUT", - ]; - - for var in &env_vars { - remove_env_var(var); - } + let keys: Vec<String> = std::env::vars() + .map(|(k, _)| k) + .filter(|k| k.starts_with("DAPI_")) + .collect(); + for k in keys { + remove_env_var(&k); + } }
69-86: Bound network‑dependent test; avoid CI flakiness.DriveClient::new() performs a real request; if upstream is unreachable it can stall. Add a short timeout or gate as an integration test.
#[tokio::test] async fn test_clients_can_be_created_with_uris() { use crate::clients::{DriveClient, TenderdashClient}; + use tokio::time::{timeout, Duration}; let config = Config::default(); - DriveClient::new(&config.dapi.drive.uri) - .await - .expect("DriveClient should be constructed even if no server is running"); + timeout(Duration::from_secs(2), DriveClient::new(&config.dapi.drive.uri)) + .await + .expect("DriveClient::new timed out") + .expect("DriveClient should be constructed even if no server is running"); TenderdashClient::new( &config.dapi.tenderdash.uri, &config.dapi.tenderdash.websocket_uri, ) .await .expect_err("TenderdashClient should fail if no server is running"); }Alternatively: #[cfg_attr(not(feature = "integration-tests"), ignore)].
330-334: Assert error category, not exact string; make the test robust.Matching "invalid digit found in string" is brittle across envy/serde versions. Assert the loader’s stable wrapper or the error variant instead.
- let error = result.expect_err("valid config").to_string(); - assert!(error.contains("invalid digit found in string")); + use crate::errors::DapiError; + let err = result.expect_err("expected invalid config to fail"); + match err { + DapiError::Configuration(msg) => { + assert!( + msg.contains("Failed to load configuration") + || msg.contains("Cannot parse config file"), + "unexpected message: {}", + msg + ); + } + other => panic!("expected configuration error, got: {}", other), + }
222-230: Prefer checking error variants over message substrings.String‑matching "Bind address '127.0.0.1:9000' must not include a port" is fragile. If possible, match the specific DapiError variant and optionally assert a stable prefix.
packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (1)
42-61: Bound the delivered sets to avoid unbounded growth
delivered_txs,delivered_blocks, anddelivered_instant_locksare unboundedHashSet<Vec<u8>>. Long‑lived streams can leak memory.
- Use LRU/TTL (e.g.,
lrucrate or time‑bucketed sets) sized to expected in‑flight window.- Record metrics and consider periodic pruning.
Also applies to: 102-147
packages/rs-dapi/src/clients/drive_client.rs (1)
80-87: Optional: includeDefaultOnEosinDriveChannelalias to match configuredon_eosYou configure
.on_eos(...)but the alias omits theOnEosgeneric. Add it for clarity (type inference likely works either way).-pub type DriveChannel = Trace< - tonic::transport::Channel, - tower_http::classify::SharedClassifier<tower_http::classify::ServerErrorsAsFailures>, - DefaultMakeSpan, - DefaultOnRequest, - DefaultOnResponse, - DefaultOnBodyChunk, ->; +pub type DriveChannel = Trace< + tonic::transport::Channel, + tower_http::classify::SharedClassifier<tower_http::classify::ServerErrorsAsFailures>, + DefaultMakeSpan, + DefaultOnRequest, + DefaultOnResponse, + DefaultOnBodyChunk, + tower_http::trace::DefaultOnEos +>;Also applies to: 144-165
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
packages/rs-dapi/src/clients/drive_client.rs(1 hunks)packages/rs-dapi/src/config/tests.rs(1 hunks)packages/rs-dapi/src/server/metrics.rs(1 hunks)packages/rs-dapi/src/services/streaming_service/block_header_stream.rs(1 hunks)packages/rs-dapi/src/services/streaming_service/transaction_stream.rs(1 hunks)packages/rs-dapi/src/services/streaming_service/zmq_listener.rs(1 hunks)
🧰 Additional context used
📓 Path-based instructions (2)
**/*.rs
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.rs: Format Rust code with cargo fmt
Run Clippy linter for Rust code
Files:
packages/rs-dapi/src/services/streaming_service/transaction_stream.rspackages/rs-dapi/src/services/streaming_service/zmq_listener.rspackages/rs-dapi/src/clients/drive_client.rspackages/rs-dapi/src/config/tests.rspackages/rs-dapi/src/services/streaming_service/block_header_stream.rspackages/rs-dapi/src/server/metrics.rs
packages/rs-*/**/*.rs
📄 CodeRabbit inference engine (AGENTS.md)
packages/rs-*/**/*.rs: Use rustfmt defaults for Rust code (format via cargo fmt --all)
Keep Rust code clippy-clean (cargo clippy)
Name Rust modules in snake_case
Use PascalCase for Rust types
Use SCREAMING_SNAKE_CASE for Rust constants
Files:
packages/rs-dapi/src/services/streaming_service/transaction_stream.rspackages/rs-dapi/src/services/streaming_service/zmq_listener.rspackages/rs-dapi/src/clients/drive_client.rspackages/rs-dapi/src/config/tests.rspackages/rs-dapi/src/services/streaming_service/block_header_stream.rspackages/rs-dapi/src/server/metrics.rs
🧠 Learnings (2)
📓 Common learnings
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/dashmate/src/test/constants/services.js:4-4
Timestamp: 2025-10-15T14:45:30.856Z
Learning: In the dashmate codebase (packages/dashmate), during the DAPI Rust migration (rs-dapi), the old service keys `dapi_api` and `dapi_core_streams` are intentionally kept in `generateEnvsFactory.js` for backward compatibility even though the test constants in `src/test/constants/services.js` have been updated to use `rs_dapi`. These deprecated keys will be removed in a future PR after the transition is complete.
📚 Learning: 2025-02-03T23:39:10.579Z
Learnt from: QuantumExplorer
PR: dashpay/platform#2450
File: packages/rs-dpp/src/data_contract/associated_token/token_perpetual_distribution/v0/methods.rs:10-12
Timestamp: 2025-02-03T23:39:10.579Z
Learning: Block interval calculations in token distribution logic should use checked arithmetic operations (checked_sub, checked_add) to prevent potential overflows, especially when dealing with block heights and intervals.
Applied to files:
packages/rs-dapi/src/services/streaming_service/block_header_stream.rs
🧬 Code graph analysis (6)
packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (2)
packages/rs-dapi/src/services/streaming_service/bloom.rs (1)
bloom_flags_from_int(103-114)packages/rs-dapi/src/services/streaming_service/mod.rs (8)
txid_bytes_from_bytes(355-363)txid_hex_from_bytes(346-352)block_hash_from_block_bytes(365-377)summarize_streaming_event(396-445)deserialize(349-349)deserialize(360-360)deserialize(368-368)short_hex(386-393)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (2)
packages/rs-dapi/src/sync.rs (5)
new(23-27)new(74-81)new(136-145)spawn(84-112)spawn(202-206)packages/rs-dapi/src/services/streaming_service/mod.rs (2)
new(48-67)summarize_zmq_event(448-489)
packages/rs-dapi/src/clients/drive_client.rs (3)
packages/rs-dapi/src/clients/tenderdash_client.rs (2)
new(53-60)new(378-417)packages/rs-dapi/src/services/streaming_service/mod.rs (1)
new(48-67)packages/rs-dapi/src/services/platform_service/mod.rs (1)
new(110-163)
packages/rs-dapi/src/config/tests.rs (3)
packages/rs-dapi/src/config/mod.rs (15)
std(225-225)default(47-54)default(136-145)default(149-153)default(157-162)default(166-174)default(198-205)load(210-216)load_from_dotenv(224-226)metrics_port(323-325)load_with_overrides(229-310)grpc_server_addr(313-315)json_rpc_addr(318-320)metrics_addr(333-341)metrics_enabled(328-330)packages/rs-dapi/src/clients/drive_client.rs (1)
new(96-132)packages/rs-dapi/src/clients/tenderdash_client.rs (2)
new(53-60)new(378-417)
packages/rs-dapi/src/services/streaming_service/block_header_stream.rs (2)
packages/rs-dapi/src/services/streaming_service/mod.rs (5)
deserialize(349-349)deserialize(360-360)deserialize(368-368)block_hash_from_block_bytes(365-377)summarize_streaming_event(396-445)packages/rs-dash-event-bus/src/event_bus.rs (2)
id(234-236)recv(239-242)
packages/rs-dapi/src/server/metrics.rs (1)
packages/rs-dapi/src/metrics.rs (1)
gather_prometheus(383-392)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (159)
- GitHub Check: Rust packages (token-history-contract) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Linting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (token-history-contract) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (token-history-contract) / Unused dependencies
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (token-history-contract) / Unused dependencies
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (token-history-contract) / Unused dependencies
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (token-history-contract) / Unused dependencies
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (token-history-contract) / Unused dependencies
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (token-history-contract) / Unused dependencies
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (token-history-contract) / Unused dependencies
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (token-history-contract) / Unused dependencies
- GitHub Check: Rust packages (wallet-utils-contract) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Linting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (dash-sdk) / Unused dependencies
- GitHub Check: Rust packages (dash-sdk) / Tests
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Unused dependencies
- GitHub Check: Rust packages (masternode-reward-shares-contract) / Formatting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Build Docker images (RS-DAPI, rs-dapi, rs-dapi) / Build RS-DAPI image
- GitHub Check: Build Docker images (DAPI, dapi, dapi) / Build DAPI image
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
🔇 Additional comments (8)
packages/rs-dapi/src/services/streaming_service/transaction_stream.rs (1)
535-543: Confirm fallback wire format for RawMerkleBlockOn merkle build failure you serialize the full
Blockand send it asRawMerkleBlock. Verify clients expect raw block bytes in this field.Do clients/specs accept raw block bytes for
responses::RawMerkleBlockon failure paths, or must this always be a serializedMerkleBlock?Also applies to: 568-576
packages/rs-dapi/src/server/metrics.rs (7)
12-43: LGTM: Clean metrics server initialization.The server setup follows standard Axum patterns with proper early-return handling for disabled metrics and optional access logging integration.
76-79: Error redaction properly implemented via conversion chain.Component errors are correctly redacted through the
From<Option<&DapiError>>→From<&DapiError>→health_error_labelchain, preventing information leakage as flagged in past reviews.
170-178: LGTM: Prometheus metrics endpoint correctly implemented.Properly delegates to
gather_prometheus()and includes appropriate content-type headers with graceful error handling.
180-227: LGTM: Well-structured health response models.Data structures use appropriate serde configuration for JSON serialization with camelCase field naming and clean optional field handling.
229-266: LGTM: Robust error redaction with exhaustive matching.The exhaustive match (no default arm) ensures all
DapiErrorvariants are explicitly mapped to generic labels, preventing information leakage while maintaining detailed trace logs for debugging.
268-306: LGTM: Safe ComponentCheck conversions with proper redaction.All conversion implementations correctly apply error redaction:
From<Option<T>>delegates to T's Into impl (which uses redaction)From<&DapiError>applieshealth_error_labelFrom<bool>uses generic error messagesThe
From<String>implementation flagged in past reviews is not present, eliminating the redaction bypass concern.
140-149: The review comment is based on an incorrect assumption and should be dismissed.The concern assumes
health.is_healthy()includes websocket connectivity in its check. However,PlatformStatusHealth.is_healthy()only checks three conditions: drive, tenderdash_status, and tenderdash_netinfo errors. Websocket connectivity is obtained independently viaplatform_service.websocket_client.is_connected()and is not part of the platform health object. The three failure counters—platform_ok,core_ok, andwebsocket_ok—represent genuinely independent health indicators with no overlap or double-counting.Likely an incorrect or invalid review comment.
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs
Outdated
Show resolved
Hide resolved
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
packages/rs-dapi/src/config/tests.rs (1)
27-40: Remove unnecessary unsafe blocks.
std::env::set_varandstd::env::remove_varare safe functions. Theunsafeblocks are unnecessary and will triggerclippy::unnecessary_unsafe, violating repository guidelines. Test serialization via#[serial]is sufficient to prevent races.As per coding guidelines.
🧹 Nitpick comments (3)
packages/dashmate/src/docker/DockerCompose.js (1)
1-14: Import reordering looks good.All imports are valid and used in the code. The functional behavior is unchanged.
For improved consistency with Airbnb style conventions, consider grouping all external imports together (currently split at line 5), then local imports:
import { Observable } from 'rxjs'; - import dockerCompose from '@dashevo/docker-compose'; import isWsl from 'is-wsl'; import hasbin from 'hasbin'; import util from 'node:util'; import semver from 'semver'; import { PACKAGE_ROOT_DIR } from '../constants.js'; import ContainerIsNotPresentError from './errors/ContainerIsNotPresentError.js'; import DockerComposeError from './errors/DockerComposeError.js'; import ServiceAlreadyRunningError from './errors/ServiceAlreadyRunningError.js'; import ServiceIsNotRunningError from './errors/ServiceIsNotRunningError.js';Or group external libraries together without internal breaks:
import { Observable } from 'rxjs'; - import dockerCompose from '@dashevo/docker-compose'; import isWsl from 'is-wsl'; - import hasbin from 'hasbin'; import util from 'node:util'; import semver from 'semver'; import { PACKAGE_ROOT_DIR } from '../constants.js'; import ContainerIsNotPresentError from './errors/ContainerIsNotPresentError.js'; import DockerComposeError from './errors/DockerComposeError.js'; import ServiceAlreadyRunningError from './errors/ServiceAlreadyRunningError.js'; import ServiceIsNotRunningError from './errors/ServiceIsNotRunningError.js';packages/rs-dapi/src/clients/tenderdash_client.rs (2)
328-334: Remove redundant Content-Type header.The
.json()method automatically sets theContent-Type: application/jsonheader, making the manual header call on line 331 redundant.Apply this diff:
let response_body = self .client .post(&self.base_url) - .header("Content-Type", "application/json") .json(request) .send()
510-517: Add documentation for subscription methods' fallback behavior.As noted in previous reviews, these subscription methods should document what happens when the WebSocket is unavailable. Based on the
TenderdashWebSocketClientimplementation, they return a broadcast receiver that will never produce events when disconnected.Consider adding doc comments like:
/// Subscribe to streaming Tenderdash transaction events. /// /// Returns a broadcast receiver for transaction events. When the WebSocket /// is not connected, the receiver will not produce any events. Use /// `is_websocket_connected()` to check connectivity before subscribing. pub fn subscribe_to_transactions(&self) -> broadcast::Receiver<TransactionEvent> {/// Subscribe to block events from Tenderdash via WebSocket. /// /// Returns a broadcast receiver for block events. When the WebSocket /// is not connected, the receiver will not produce any events. Use /// `is_websocket_connected()` to check connectivity before subscribing. pub fn subscribe_to_blocks(&self) -> broadcast::Receiver<BlockEvent> {Based on learnings
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
packages/dashmate/src/docker/DockerCompose.js(1 hunks)packages/rs-dapi/src/clients/tenderdash_client.rs(1 hunks)packages/rs-dapi/src/config/tests.rs(1 hunks)
🧰 Additional context used
📓 Path-based instructions (3)
**/*.rs
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.rs: Format Rust code with cargo fmt
Run Clippy linter for Rust code
Files:
packages/rs-dapi/src/config/tests.rspackages/rs-dapi/src/clients/tenderdash_client.rs
packages/rs-*/**/*.rs
📄 CodeRabbit inference engine (AGENTS.md)
packages/rs-*/**/*.rs: Use rustfmt defaults for Rust code (format via cargo fmt --all)
Keep Rust code clippy-clean (cargo clippy)
Name Rust modules in snake_case
Use PascalCase for Rust types
Use SCREAMING_SNAKE_CASE for Rust constants
Files:
packages/rs-dapi/src/config/tests.rspackages/rs-dapi/src/clients/tenderdash_client.rs
packages/**/**/*.{js,ts,jsx,tsx}
📄 CodeRabbit inference engine (AGENTS.md)
packages/**/**/*.{js,ts,jsx,tsx}: Adhere to ESLint with Airbnb/TypeScript configs for JS/TS code
Use camelCase for JS/TS variables and functions
Use PascalCase for JS/TS classes
Prefer kebab-case filenames within JS packages
Files:
packages/dashmate/src/docker/DockerCompose.js
🧠 Learnings (3)
📓 Common learnings
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/dashmate/src/test/constants/services.js:4-4
Timestamp: 2025-10-15T14:45:30.856Z
Learning: In the dashmate codebase (packages/dashmate), during the DAPI Rust migration (rs-dapi), the old service keys `dapi_api` and `dapi_core_streams` are intentionally kept in `generateEnvsFactory.js` for backward compatibility even though the test constants in `src/test/constants/services.js` have been updated to use `rs_dapi`. These deprecated keys will be removed in a future PR after the transition is complete.
📚 Learning: 2025-10-09T15:59:33.162Z
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/rs-dapi/src/clients/tenderdash_client.rs:0-0
Timestamp: 2025-10-09T15:59:33.162Z
Learning: In the rs-dapi codebase, the `Workers::spawn` method (in `packages/rs-dapi/src/sync.rs`) accepts futures that return `Result<O, E>`, not `()`. Its signature is `fn spawn<F, O, E>(&self, fut: F) where F: Future<Output = Result<O, E>> + Send + 'static`. The method handles error logging internally, so spawned tasks should return Results directly without additional error handling wrappers.
Applied to files:
packages/rs-dapi/src/clients/tenderdash_client.rs
📚 Learning: 2025-10-09T15:59:41.943Z
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/rs-dapi/src/clients/tenderdash_client.rs:0-0
Timestamp: 2025-10-09T15:59:41.943Z
Learning: In packages/rs-dapi, the Workers::spawn method accepts futures that return Result<O, E> where E: Into<DapiError>. It handles error logging internally, so spawned tasks should return Result types rather than handling errors inside the task. The Workers implementation automatically logs worker failures.
Applied to files:
packages/rs-dapi/src/clients/tenderdash_client.rs
🧬 Code graph analysis (2)
packages/rs-dapi/src/config/tests.rs (3)
packages/rs-dapi/src/config/mod.rs (15)
std(225-225)default(47-54)default(136-145)default(149-153)default(157-162)default(166-174)default(198-205)load(210-216)load_from_dotenv(224-226)metrics_port(323-325)load_with_overrides(229-310)grpc_server_addr(313-315)json_rpc_addr(318-320)metrics_addr(333-341)metrics_enabled(328-330)packages/rs-dapi/src/clients/tenderdash_client.rs (2)
new(53-60)new(379-422)packages/rs-dapi/src/clients/drive_client.rs (1)
new(96-132)
packages/rs-dapi/src/clients/tenderdash_client.rs (3)
packages/rs-dapi/src/utils.rs (3)
deserialize_string_number_or_null(147-163)deserialize_string_or_number(23-86)generate_jsonrpc_id(12-21)packages/rs-dapi/src/clients/tenderdash_websocket.rs (2)
new(78-88)test_connection(106-125)packages/rs-dapi/src/error.rs (2)
from_tenderdash_error(170-172)client(201-203)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (148)
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (drive-abci) / Unused dependencies
- GitHub Check: Rust packages (drive-abci) / Linting
- GitHub Check: Rust packages (drive-abci) / Formatting
- GitHub Check: Rust packages (drive-abci) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Formatting
- GitHub Check: Rust packages (withdrawals-contract) / Unused dependencies
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (drive-abci) / Unused dependencies
- GitHub Check: Rust packages (drive-abci) / Linting
- GitHub Check: Rust packages (drive-abci) / Formatting
- GitHub Check: Rust packages (drive-abci) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Formatting
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (drive-abci) / Unused dependencies
- GitHub Check: Rust packages (drive-abci) / Linting
- GitHub Check: Rust packages (drive-abci) / Formatting
- GitHub Check: Rust packages (drive-abci) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Formatting
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (dash-sdk) / Detect immutable structure changes
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (drive-abci) / Unused dependencies
- GitHub Check: Rust packages (drive-abci) / Linting
- GitHub Check: Rust packages (drive-abci) / Formatting
- GitHub Check: Rust packages (drive-abci) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Formatting
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (dash-sdk) / Detect immutable structure changes
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (drive-abci) / Unused dependencies
- GitHub Check: Rust packages (drive-abci) / Linting
- GitHub Check: Rust packages (drive-abci) / Formatting
- GitHub Check: Rust packages (drive-abci) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Formatting
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (dash-sdk) / Detect immutable structure changes
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (drive-abci) / Unused dependencies
- GitHub Check: Rust packages (drive-abci) / Linting
- GitHub Check: Rust packages (drive-abci) / Formatting
- GitHub Check: Rust packages (drive-abci) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Formatting
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (dash-sdk) / Detect immutable structure changes
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (drive-abci) / Unused dependencies
- GitHub Check: Rust packages (drive-abci) / Linting
- GitHub Check: Rust packages (drive-abci) / Formatting
- GitHub Check: Rust packages (drive-abci) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Formatting
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (dash-sdk) / Detect immutable structure changes
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (drive-abci) / Unused dependencies
- GitHub Check: Rust packages (drive-abci) / Linting
- GitHub Check: Rust packages (drive-abci) / Formatting
- GitHub Check: Rust packages (drive-abci) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Formatting
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (dash-sdk) / Detect immutable structure changes
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (drive-abci) / Unused dependencies
- GitHub Check: Rust packages (drive-abci) / Linting
- GitHub Check: Rust packages (drive-abci) / Formatting
- GitHub Check: Rust packages (drive-abci) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Unused dependencies
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Formatting
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
- GitHub Check: Rust packages (dash-sdk) / Detect immutable structure changes
- GitHub Check: Rust packages (dash-sdk) / Formatting
- GitHub Check: Rust packages (dash-sdk) / Check each feature
- GitHub Check: Rust packages (drive-abci) / Check each feature
- GitHub Check: Rust packages (drive-abci) / Unused dependencies
- GitHub Check: Rust packages (drive-abci) / Linting
- GitHub Check: Rust packages (drive-abci) / Formatting
- GitHub Check: Rust packages (drive-abci) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Tests
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Linting
- GitHub Check: Rust packages (json-schema-compatibility-validator) / Formatting
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Build Docker images (Drive, drive, drive-abci) / Build Drive image
- GitHub Check: Build Docker images (Dashmate helper, dashmate-helper, dashmate-helper) / Build Dashmate helper image
🔇 Additional comments (11)
packages/rs-dapi/src/config/tests.rs (6)
8-25: LGTM!The cleanup helper covers all configuration environment variables and is used appropriately in serialized tests.
42-86: LGTM!The default config, environment override, and async client creation tests are well-structured. The async test correctly validates that clients can be constructed even when upstream services are unavailable (degraded mode).
88-176: LGTM!The dotenv loading tests comprehensively cover:
- Full configuration loading from file
- Partial loading with default fallback
- Error handling for nonexistent files
All tests properly use
#[serial]and cleanup helpers.
178-245: LGTM!Address resolution tests properly validate:
- IPv4 and IPv6 literal parsing
- Hostname resolution to loopback
- Rejection of ports in bind addresses
- Invalid address error handling
247-310: LGTM!Override precedence tests correctly validate the layered configuration system (dotenv < env < CLI), ensuring CLI overrides have highest priority.
339-422: LGTM!The remaining tests comprehensively validate:
- Socket address formatting for all server endpoints
- Custom bind address handling
- Metrics disabling when port is zero
- Default config validation success
- Validation failure for invalid bind addresses
All assertions are clear and cover both positive and negative cases.
packages/rs-dapi/src/clients/tenderdash_client.rs (5)
18-34: Well-documented client structure with clear error-handling architecture.The struct documentation clearly defines the error-handling philosophy (technical vs service errors) and tracing behavior, which will help maintainers understand logging expectations.
379-422: Excellent implementation with timeouts and tracing middleware.This method properly addresses previous review concerns by:
- Configuring connection and request timeouts (lines 387-388)
- Adding
TracingMiddlewarefor observability (line 396)- Implementing graceful degradation when connectivity validation fails
The degraded-mode approach (continuing despite validation failures while logging warnings) is a sensible design choice for service resilience.
452-466: Verify the error-masking design innet_info.This method catches all errors from the RPC call and returns
NetInfoResponse::default()instead of propagating them. While this may be intentional for resilience (e.g., allowing health checks to continue with default values), it prevents callers from distinguishing between a successful response with zero peers versus a connection failure.Consider documenting this behavior in the method's doc comment, or evaluate whether callers should handle the error explicitly.
87-316: Robust data models with flexible deserialization.The extensive use of custom deserializers (
deserialize_string_or_number,deserialize_string_number_or_null) combined with#[serde(default)]makes these models resilient to variations in Tenderdash's JSON-RPC responses. This is a good defensive approach for external API integration.
443-509: Clean and consistent RPC method implementations.All RPC methods follow a uniform pattern (construct params → create request → delegate to
post), making the codebase maintainable and easy to extend with additional RPC endpoints.
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (5)
packages/rs-dapi/src/clients/drive_client.rs (1)
170-178: Fix compilation error: cannot move out of borrowed reference.Line 177 dereferences
*requestwhererequestis&GetStatusRequest. SinceGetStatusRequest(a protobuf message type) is notCopy, this attempts to move out of a shared reference and will fail to compile.Apply this diff to clone the request:
pub async fn get_drive_status( &self, request: &GetStatusRequest, ) -> Result<DriveStatusResponse, tonic::Status> { trace!("Making get_status gRPC call to Drive"); // Make gRPC call to Drive with timing - let drive_response = self.get_client().get_status(*request).await?.into_inner(); + let drive_response = self + .get_client() + .get_status(request.clone()) + .await? + .into_inner();Note: This issue was flagged in a previous review but remains unresolved in the current code.
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (4)
616-644: Re-evaluate “ping” subscribe/unsubscribe health check.This is still an indirect, potentially non‑idempotent probe. With
last_recv_*monotonic tracking in place, consider removing this fallback or guard it behind a debug/config flag.- let current_status = self.socket.subscribe("ping").await.is_ok(); - // Unsubscribe immediately to avoid resource waste - self.socket - .unsubscribe("ping") - .await - .inspect_err(|e| { - debug!(error = %e, "Error unsubscribing from ping topic during health check"); - }) - .ok(); + #[cfg(feature = "zmq_health_ping")] + let current_status = { + let ok = self.socket.subscribe("ping").await.is_ok(); + let _ = self.socket.unsubscribe("ping").await; + ok + }; + #[cfg(not(feature = "zmq_health_ping"))] + let current_status = false; // rely solely on last_recv; no active probeIf you need an active probe, prefer a lightweight socket op exposed by the library or Core RPC.
379-382: Docstring mismatches semantics. Rename doc or method accordingly.Method is
is_running()but doc says “connected”. It checks the cancellation token only.- /// Check if the ZMQ listener is connected (placeholder) + /// Returns true while the listener task is running (not cancelled).Optionally add a separate
is_socket_connected()that reads theconnectedflag fromZmqConnectionif you need actual socket state.
523-539: Don’t treat malformed rawtxlocksig payload as “lock bytes”. Drop it.Returning
(None, Some(data))when tx decode fails misleads downstream (lock bytes position unknown). Return(None, None)and handle it upstream.- Err(_) => (None, Some(data)), + Err(e) => { + debug!(error = ?e, data_len = data.len(), "Failed to decode transaction from rawtxlocksig payload"); + (None, None) + }
491-508: Handle(None, None)from split_tx_and_lock and drop malformed events.Pairs with the previous fix.
- let (tx_bytes, lock_bytes_opt) = split_tx_and_lock(data); + let (tx_bytes, lock_bytes_opt) = split_tx_and_lock(data); + if tx_bytes.is_none() && lock_bytes_opt.is_none() { + debug!("rawtxlocksig payload has malformed transaction data, dropping event"); + return None; + } if let Some(lock_bytes) = lock_bytes_opt && !lock_bytes.is_empty() { Some(ZmqEvent::RawTransactionLock { tx_bytes, lock_bytes, }) } else { debug!("rawtxlocksig payload missing instant lock bytes"); None }
🧹 Nitpick comments (5)
packages/rs-dapi/src/clients/drive_client.rs (2)
38-78: Consider documenting Option semantics for data models.All fields in these data model structs are
Option<T>, but there's no documentation explaining when fields would beNone. Adding doc comments would help consumers understand whetherNoneindicates:
- Missing data from the Drive service
- Optional features/fields in certain protocol versions
- Potential errors during response parsing
Example documentation pattern:
+/// Drive status information returned by the Drive service. +/// +/// All fields are optional as they may be absent depending on Drive version +/// or the service's current state. #[derive(Debug, Serialize, Deserialize, Default)] pub struct DriveStatusResponse { + /// Version information (software and protocol versions) pub version: Option<DriveVersion>, + /// Blockchain state information pub chain: Option<DriveChain>, + /// Timing information (block time, genesis, epoch) pub time: Option<DriveTime>, }
248-273: Consider expanding test coverage beyond smoke test.The current test only verifies that
DriveClient::newdoesn't panic. Consider adding:
- Mock server tests to validate
get_drive_statusresponse mapping- Tests for error cases (invalid URI, malformed responses)
- Programmatic tracing validation using
tracing-subscriber-testor similarExample test structure using a mock server:
#[tokio::test] async fn test_get_drive_status_maps_response_correctly() { // Set up mock gRPC server returning known DriveStatus // Call client.get_drive_status(...) // Assert mapped fields match expected values } #[tokio::test] async fn test_invalid_uri_returns_error() { let result = DriveClient::new("not a valid uri").await; assert!(result.is_err()); }packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (3)
425-433: Backoff comment says “with jitter” but code has none.Either add jitter or fix the comment. Jitter avoids thundering herds on restarts.
- // Exponential backoff with jitter, capped at 300 seconds - delay = std::cmp::min(delay * 2, Duration::from_secs(300)); + // Exponential backoff with jitter, capped at 300 seconds + let jitter_ms = fastrand::u64(..500); + delay = std::cmp::min(delay * 2 + Duration::from_millis(jitter_ms), Duration::from_secs(300));Note: add dependency fastrand = "2". If avoiding deps, use a simple LCG seeded once.
492-495: Avoid hex-encoding full payload in hot path.
hex::encode(&data)is expensive and allocates; trace-level still evaluates args. Log a short prefix.- tracing::trace!( - data = hex::encode(&data), - "Parsing rawtxlocksig ZMQ message" - ); + tracing::trace!(prefix = %crate::utils::short_hex(&data, 12), len = data.len(), "Parsing rawtxlocksig ZMQ message");If short_hex isn’t available here, inline a small helper.
374-377: subscribe() can’t fail; simplify signature.
broadcast::Sender::subscribe()is infallible. Consider returning the receiver directly to reduce error plumbing.- pub fn subscribe(&self) -> DAPIResult<broadcast::Receiver<ZmqEvent>> { - Ok(self.event_sender.subscribe()) - } + pub fn subscribe(&self) -> broadcast::Receiver<ZmqEvent> { + self.event_sender.subscribe() + }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (2)
packages/rs-dapi/src/clients/drive_client.rs(1 hunks)packages/rs-dapi/src/services/streaming_service/zmq_listener.rs(1 hunks)
🧰 Additional context used
📓 Path-based instructions (2)
**/*.rs
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.rs: Format Rust code with cargo fmt
Run Clippy linter for Rust code
Files:
packages/rs-dapi/src/clients/drive_client.rspackages/rs-dapi/src/services/streaming_service/zmq_listener.rs
packages/rs-*/**/*.rs
📄 CodeRabbit inference engine (AGENTS.md)
packages/rs-*/**/*.rs: Use rustfmt defaults for Rust code (format via cargo fmt --all)
Keep Rust code clippy-clean (cargo clippy)
Name Rust modules in snake_case
Use PascalCase for Rust types
Use SCREAMING_SNAKE_CASE for Rust constants
Files:
packages/rs-dapi/src/clients/drive_client.rspackages/rs-dapi/src/services/streaming_service/zmq_listener.rs
🧠 Learnings (1)
📓 Common learnings
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/dashmate/src/test/constants/services.js:4-4
Timestamp: 2025-10-15T14:45:30.856Z
Learning: In the dashmate codebase (packages/dashmate), during the DAPI Rust migration (rs-dapi), the old service keys `dapi_api` and `dapi_core_streams` are intentionally kept in `generateEnvsFactory.js` for backward compatibility even though the test constants in `src/test/constants/services.js` have been updated to use `rs_dapi`. These deprecated keys will be removed in a future PR after the transition is complete.
🧬 Code graph analysis (2)
packages/rs-dapi/src/clients/drive_client.rs (2)
packages/rs-dapi/src/config/mod.rs (6)
default(47-54)default(136-145)default(149-153)default(157-162)default(166-174)default(198-205)packages/rs-dapi/src/clients/tenderdash_client.rs (2)
new(53-60)new(379-422)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (2)
packages/rs-dapi/src/services/streaming_service/mod.rs (2)
new(48-67)summarize_zmq_event(448-489)packages/rs-dapi/src/sync.rs (6)
new(23-27)new(74-81)new(136-145)default(117-119)spawn(84-112)spawn(202-206)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: Swift SDK and Example build (warnings as errors)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
🔇 Additional comments (3)
packages/rs-dapi/src/clients/drive_client.rs (3)
89-132: Constructor design is sound.The degraded-mode approach (logging validation failures but returning
Ok) aligns with the TenderdashClient pattern and allows the service to start even when Drive is temporarily unavailable. Message size limits and compression handling are appropriate.
134-168: Tracing setup is comprehensive and well-configured.The lazy channel creation, detailed trace configuration, and error normalization provide robust observability and error handling for Drive connectivity.
195-204: No issues found—let-chain syntax is fully supported by the project's MSRV.The project specifies
rust-version = "1.89"in the root Cargo.toml, which exceeds the 1.64 minimum required for let-chain syntax stabilization. The code on lines 195-196 is fully compatible with the project's minimum supported Rust version.
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (1)
381-384: Docstring mismatch: this checks “running”, not “connected”.Update the comment to reflect cancellation state, not socket connectivity. This was noted previously.
- /// Check if the ZMQ listener is connected (placeholder) + /// Check if the ZMQ listener task is still running (not cancelled) pub fn is_running(&self) -> bool { !self.cancel.is_cancelled() }
🧹 Nitpick comments (1)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (1)
433-435: Comment says “with jitter” but code has no jitter.Either add jitter or fix the comment. Given “chill” scope, suggest correcting the comment now; optionally add jitter later.
Option A (comment-only):
- // Exponential backoff with jitter, capped at 300 seconds + // Exponential backoff (no jitter), capped at 300 seconds delay = std::cmp::min(delay * 2, Duration::from_secs(300));Option B (implement jitter; requires a RNG dependency like
randorfastrandin Cargo.toml):- delay = std::cmp::min(delay * 2, Duration::from_secs(300)); + let base = std::cmp::min(delay * 2, Duration::from_secs(300)); + // add ±10% jitter + let jitter = (base.as_millis() as i128) / 10; + let offset_ms = (fastrand::i128(-jitter..=jitter)) as i64; + let jittered = (base.as_millis() as i128 + offset_ms as i128).max(0) as u64; + delay = Duration::from_millis(jittered);
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs(1 hunks)
🧰 Additional context used
📓 Path-based instructions (2)
**/*.rs
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.rs: Format Rust code with cargo fmt
Run Clippy linter for Rust code
Files:
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs
packages/rs-*/**/*.rs
📄 CodeRabbit inference engine (AGENTS.md)
packages/rs-*/**/*.rs: Use rustfmt defaults for Rust code (format via cargo fmt --all)
Keep Rust code clippy-clean (cargo clippy)
Name Rust modules in snake_case
Use PascalCase for Rust types
Use SCREAMING_SNAKE_CASE for Rust constants
Files:
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs
🧠 Learnings (1)
📓 Common learnings
Learnt from: lklimek
PR: dashpay/platform#2716
File: packages/dashmate/src/test/constants/services.js:4-4
Timestamp: 2025-10-15T14:45:30.856Z
Learning: In the dashmate codebase (packages/dashmate), during the DAPI Rust migration (rs-dapi), the old service keys `dapi_api` and `dapi_core_streams` are intentionally kept in `generateEnvsFactory.js` for backward compatibility even though the test constants in `src/test/constants/services.js` have been updated to use `rs_dapi`. These deprecated keys will be removed in a future PR after the transition is complete.
🧬 Code graph analysis (1)
packages/rs-dapi/src/services/streaming_service/zmq_listener.rs (3)
packages/rs-dapi/src/sync.rs (6)
new(23-27)new(74-81)new(136-145)default(117-119)spawn(84-112)spawn(202-206)packages/rs-dapi/src/services/streaming_service/mod.rs (2)
new(48-67)summarize_zmq_event(448-489)packages/rs-dapi/src/error.rs (1)
timeout(251-253)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios-sim)
- GitHub Check: Build rs-sdk-ffi for iOS targets (aarch64-apple-ios)
- GitHub Check: Swift SDK and Example build (warnings as errors)
| if let Some(error) = first_error { | ||
| debug!( | ||
| ?error, | ||
| "ZMQ subscription errors occured, trying to unsubscribe from successful topics", | ||
| ); | ||
|
|
||
| if let Err(unsub_err) = self.zmq_unsubscribe_all(socket).await { | ||
| debug!(error = %unsub_err, "Unsubscribe during rollback failed; preserving original subscribe error"); | ||
| } |
There was a problem hiding this comment.
Typo in log message (“occured”).
Fix the spelling to “occurred”.
- "ZMQ subscription errors occured, trying to unsubscribe from successful topics",
+ "ZMQ subscription errors occurred, trying to unsubscribe from successful topics",🤖 Prompt for AI Agents
In packages/rs-dapi/src/services/streaming_service/zmq_listener.rs around lines
193 to 201, the debug log message contains a typo ("occured"); update the log
string to "occurred" so the message reads "ZMQ subscription errors occurred,
trying to unsubscribe from successful topics" and keep the surrounding debug
context and formatting unchanged.
2 participants
Issue being fixed or feature implemented
What was done?
How Has This Been Tested?
Breaking Changes
Checklist:
For repository code-owners and collaborators only
Summary by CodeRabbit
New Features
Infrastructure
Documentation
Tests
Chores